실제로 시스템을 구축하기 전에 필요한 것들을 준비해보자. 먼저 개발 환경부터 세팅해야 해.

📦 필요한 소프트웨어

🐘 PHP rdkafka 확장 설치

PHP에서 Kafka를 사용하려면 rdkafka 확장을 설치해야 해. 이건 librdkafka C 라이브러리의 PHP 바인딩이야. 성능이 정말 좋아서 프로덕션 환경에서도 충분히 사용할 수 있어! 🚀

# librdkafka 설치
sudo apt-get install librdkafka-dev

# PECL로 PHP 확장 설치
sudo pecl install rdkafka

# php.ini에 확장 추가
echo "extension=rdkafka.so" | sudo tee -a /etc/php/8.1/cli/php.ini
echo "extension=rdkafka.so" | sudo tee -a /etc/php/8.1/fpm/php.ini

# 설치 확인
php -m | grep rdkafka

🎵 Kafka 설치 및 실행

개발 환경에서는 Docker를 사용하면 정말 편해. Docker Compose로 Kafka와 Zookeeper를 한 번에 띄울 수 있거든!

# docker-compose.yml
version: '3'
services:
  zookeeper:
    image: confluentinc/cp-zookeeper:latest
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181
      ZOOKEEPER_TICK_TIME: 2000
    ports:
      - "2181:2181"

  kafka:
    image: confluentinc/cp-kafka:latest
    depends_on:
      - zookeeper
    ports:
      - "9092:9092"
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_AUTO_CREATE_TOPICS_ENABLE: "true"

이제 docker-compose up -d 명령어로 실행하면 끝! 정말 간단하지? 😎

자, 이제 본격적으로 코드를 작성해보자! 먼저 로그를 Kafka로 전송하는 Producer부터 만들어볼게.

🎨 기본 Producer 클래스

재사용 가능한 Producer 클래스를 만들면 여러 곳에서 편하게 사용할 수 있어. 객체지향적으로 설계해보자!

<?php
// src/KafkaProducer.php

namespace App\Kafka;

use RdKafka\Conf;
use RdKafka\Producer;
use RdKafka\ProducerTopic;

class KafkaProducer
{
    private Producer $producer;
    private array $topics = [];
    private array $config;

    public function __construct(array $config = [])
    {
        $this->config = array_merge([
            'metadata.broker.list' => 'localhost:9092',
            'compression.codec' => 'snappy',
            'queue.buffering.max.messages' => 100000,
            'queue.buffering.max.ms' => 100,
            'batch.num.messages' => 10000,
        ], $config);

        $this->initProducer();
    }

    private function initProducer(): void
    {
        $conf = new Conf();
        
        foreach ($this->config as $key => $value) {
            $conf->set($key, $value);
        }

        // 에러 콜백 설정
        $conf->setErrorCb(function ($kafka, $err, $reason) {
            error_log(sprintf(
                "Kafka error: %s (reason: %s)",
                rd_kafka_err2str($err),
                $reason
            ));
        });

        // 전송 완료 콜백 설정
        $conf->setDrMsgCb(function ($kafka, $message) {
            if ($message->err) {
                error_log(sprintf(
                    "Message delivery failed: %s",
                    $message->errstr()
                ));
            }
        });

        $this->producer = new Producer($conf);
    }

    public function getTopic(string $topicName): ProducerTopic
    {
        if (!isset($this->topics[$topicName])) {
            $this->topics[$topicName] = $this->producer->newTopic($topicName);
        }

        return $this->topics[$topicName];
    }

    public function send(
        string $topicName,
        string $message,
        ?string $key = null,
        ?int $partition = RD_KAFKA_PARTITION_UA
    ): bool {
        try {
            $topic = $this->getTopic($topicName);
            
            $topic->produce(
                $partition,
                0,
                $message,
                $key
            );

            // 비동기 전송을 위해 poll 호출
            $this->producer->poll(0);

            return true;
        } catch (\Exception $e) {
            error_log("Failed to send message: " . $e->getMessage());
            return false;
        }
    }

    public function sendBatch(string $topicName, array $messages): int
    {
        $successCount = 0;
        $topic = $this->getTopic($topicName);

        foreach ($messages as $message) {
            try {
                $key = $message['key'] ?? null;
                $payload = $message['payload'] ?? $message;

                if (is_array($payload)) {
                    $payload = json_encode($payload);
                }

                $topic->produce(
                    RD_KAFKA_PARTITION_UA,
                    0,
                    $payload,
                    $key
                );

                $successCount++;

                // 주기적으로 poll 호출
                if ($successCount % 100 === 0) {
                    $this->producer->poll(0);
                }
            } catch (\Exception $e) {
                error_log("Failed to send batch message: " . $e->getMessage());
            }
        }

        // 남은 메시지 전송 완료 대기
        $this->flush();

        return $successCount;
    }

    public function flush(int $timeoutMs = 10000): void
    {
        $result = $this->producer->flush($timeoutMs);
        
        if (RD_KAFKA_RESP_ERR_NO_ERROR !== $result) {
            error_log("Flush failed with error: " . rd_kafka_err2str($result));
        }
    }

    public function __destruct()
    {
        $this->flush();
    }
}

와! 코드가 좀 길지? 하지만 하나하나 뜯어보면 별거 없어. 핵심은 비동기 전송과 배치 처리야. 🎯

📝 로그 수집 래퍼 클래스

이제 실제로 애플리케이션에서 사용할 로그 수집 클래스를 만들어보자. 이 클래스는 다양한 타입의 로그를 쉽게 전송할 수 있게 해줄 거야.

<?php
// src/LogCollector.php

namespace App\Kafka;

class LogCollector
{
    private KafkaProducer $producer;
    private string $applicationName;
    private string $environment;

    public function __construct(
        KafkaProducer $producer,
        string $applicationName = 'my-app',
        string $environment = 'production'
    ) {
        $this->producer = $producer;
        $this->applicationName = $applicationName;
        $this->environment = $environment;
    }

    public function info(string $message, array $context = []): void
    {
        $this->log('info', $message, $context);
    }

    public function warning(string $message, array $context = []): void
    {
        $this->log('warning', $message, $context);
    }

    public function error(string $message, array $context = []): void
    {
        $this->log('error', $message, $context);
    }

    public function critical(string $message, array $context = []): void
    {
        $this->log('critical', $message, $context);
    }

    public function userAction(string $userId, string $action, array $data = []): void
    {
        $logData = [
            'type' => 'user_action',
            'user_id' => $userId,
            'action' => $action,
            'data' => $data,
            'timestamp' => microtime(true),
            'application' => $this->applicationName,
            'environment' => $this->environment,
            'server' => gethostname(),
            'ip' => $_SERVER['REMOTE_ADDR'] ?? 'unknown',
            'user_agent' => $_SERVER['HTTP_USER_AGENT'] ?? 'unknown',
        ];

        $this->producer->send(
            'user-actions',
            json_encode($logData),
            $userId
        );
    }

    public function apiCall(
        string $endpoint,
        string $method,
        int $statusCode,
        float $duration,
        array $context = []
    ): void {
        $logData = [
            'type' => 'api_call',
            'endpoint' => $endpoint,
            'method' => $method,
            'status_code' => $statusCode,
            'duration' => $duration,
            'context' => $context,
            'timestamp' => microtime(true),
            'application' => $this->applicationName,
            'environment' => $this->environment,
        ];

        $this->producer->send(
            'api-logs',
            json_encode($logData)
        );
    }

    public function performance(string $operation, float $duration, array $metrics = []): void
    {
        $logData = [
            'type' => 'performance',
            'operation' => $operation,
            'duration' => $duration,
            'metrics' => array_merge($metrics, [
                'memory_usage' => memory_get_usage(true),
                'memory_peak' => memory_get_peak_usage(true),
            ]),
            'timestamp' => microtime(true),
            'application' => $this->applicationName,
            'environment' => $this->environment,
        ];

        $this->producer->send(
            'performance-logs',
            json_encode($logData)
        );
    }

    private function log(string $level, string $message, array $context = []): void
    {
        $logData = [
            'level' => $level,
            'message' => $message,
            'context' => $context,
            'timestamp' => microtime(true),
            'application' => $this->applicationName,
            'environment' => $this->environment,
            'file' => $context['file'] ?? null,
            'line' => $context['line'] ?? null,
            'trace' => $context['trace'] ?? null,
        ];

        $topicMap = [
            'info' => 'info-logs',
            'warning' => 'warning-logs',
            'error' => 'error-logs',
            'critical' => 'critical-logs',
        ];

        $topic = $topicMap[$level] ?? 'general-logs';

        $this->producer->send(
            $topic,
            json_encode($logData)
        );
    }
}

이제 애플리케이션에서 정말 쉽게 로그를 수집할 수 있어! 사용 예시를 볼까? 😊

<?php
// 사용 예시

$producer = new KafkaProducer([
    'metadata.broker.list' => 'localhost:9092',
]);

$logger = new LogCollector($producer, 'my-awesome-app', 'production');

// 일반 로그
$logger->info('User logged in successfully', [
    'user_id' => 12345,
    'login_method' => 'oauth',
]);

// 사용자 액션 로그
$logger->userAction('user-12345', 'purchase', [
    'product_id' => 'prod-789',
    'amount' => 29.99,
    'currency' => 'USD',
]);

// API 호출 로그
$startTime = microtime(true);
// ... API 처리 ...
$duration = microtime(true) - $startTime;

$logger->apiCall(
    '/api/v1/users',
    'GET',
    200,
    $duration,
    ['query_params' => $_GET]
);

// 성능 로그
$logger->performance('database_query', 0.125, [
    'query' => 'SELECT * FROM users WHERE id = ?',
    'rows_affected' => 1,
]);

정말 간단하지? 이렇게 하면 애플리케이션의 모든 중요한 이벤트를 Kafka로 전송할 수 있어. 그리고 나중에 이 데이터를 분석해서 사용자 행동 패턴을 파악하거나, 성능 병목을 찾거나, 에러를 모니터링할 수 있지! 📊

Producer를 만들었으니 이제 Consumer를 만들 차례야! Consumer는 Kafka에서 메시지를 읽어와서 실제로 처리하는 역할을 해. 여기서 중요한 건 안정성과 확장성이야. 🎪

🎯 기본 Consumer 클래스

<?php
// src/KafkaConsumer.php

namespace App\Kafka;

use RdKafka\Conf;
use RdKafka\KafkaConsumer;
use RdKafka\Message;
use RdKafka\TopicPartition;

class Consumer
{
    private KafkaConsumer $consumer;
    private array $config;
    private bool $running = true;
    private array $handlers = [];

    public function __construct(array $config = [])
    {
        $this->config = array_merge([
            'metadata.broker.list' => 'localhost:9092',
            'group.id' => 'default-consumer-group',
            'auto.offset.reset' => 'earliest',
            'enable.auto.commit' => 'false',
            'compression.codec' => 'snappy',
        ], $config);

        $this->initConsumer();
    }

    private function initConsumer(): void
    {
        $conf = new Conf();

        foreach ($this->config as $key => $value) {
            $conf->set($key, $value);
        }

        // 리밸런싱 콜백
        $conf->setRebalanceCb(function (
            KafkaConsumer $kafka,
            $err,
            array $partitions = null
        ) {
            switch ($err) {
                case RD_KAFKA_RESP_ERR__ASSIGN_PARTITIONS:
                    echo "Partitions assigned: " . count($partitions) . "\n";
                    $kafka->assign($partitions);
                    break;

                case RD_KAFKA_RESP_ERR__REVOKE_PARTITIONS:
                    echo "Partitions revoked\n";
                    $kafka->assign(null);
                    break;

                default:
                    error_log("Rebalance error: " . rd_kafka_err2str($err));
                    break;
            }
        });

        // 에러 콜백
        $conf->setErrorCb(function ($kafka, $err, $reason) {
            error_log(sprintf(
                "Consumer error: %s (reason: %s)",
                rd_kafka_err2str($err),
                $reason
            ));
        });

        $this->consumer = new KafkaConsumer($conf);
    }

    public function subscribe(array $topics): void
    {
        $this->consumer->subscribe($topics);
        echo "Subscribed to topics: " . implode(', ', $topics) . "\n";
    }

    public function registerHandler(string $topic, callable $handler): void
    {
        $this->handlers[$topic] = $handler;
    }

    public function consume(int $timeoutMs = 1000): void
    {
        echo "Starting consumer...\n";

        // 시그널 핸들러 등록
        pcntl_signal(SIGTERM, [$this, 'shutdown']);
        pcntl_signal(SIGINT, [$this, 'shutdown']);

        while ($this->running) {
            pcntl_signal_dispatch();

            $message = $this->consumer->consume($timeoutMs);

            switch ($message->err) {
                case RD_KAFKA_RESP_ERR_NO_ERROR:
                    $this->handleMessage($message);
                    break;

                case RD_KAFKA_RESP_ERR__PARTITION_EOF:
                    // 파티션 끝에 도달 (정상)
                    break;

                case RD_KAFKA_RESP_ERR__TIMED_OUT:
                    // 타임아웃 (정상)
                    break;

                default:
                    error_log("Consumer error: " . $message->errstr());
                    break;
            }
        }

        echo "Consumer stopped\n";
    }

    private function handleMessage(Message $message): void
    {
        $topic = $message->topic_name;
        
        try {
            $payload = json_decode($message->payload, true);

            if (json_last_error() !== JSON_ERROR_NONE) {
                throw new \Exception("Invalid JSON: " . json_last_error_msg());
            }

            // 토픽별 핸들러 실행
            if (isset($this->handlers[$topic])) {
                $handler = $this->handlers[$topic];
                $handler($payload, $message);
            } else {
                // 기본 핸들러
                $this->defaultHandler($payload, $message);
            }

            // 수동 커밋
            $this->consumer->commit($message);

        } catch (\Exception $e) {
            error_log(sprintf(
                "Failed to handle message from topic %s: %s",
                $topic,
                $e->getMessage()
            ));

            // 에러 처리 로직 (재시도, DLQ 전송 등)
            $this->handleError($message, $e);
        }
    }

    private function defaultHandler(array $payload, Message $message): void
    {
        echo sprintf(
            "[%s] Received message: %s\n",
            $message->topic_name,
            json_encode($payload, JSON_PRETTY_PRINT)
        );
    }

    private function handleError(Message $message, \Exception $e): void
    {
        // Dead Letter Queue로 전송하거나 별도 에러 로그 저장
        error_log(sprintf(
            "Error handling message: %s\nPayload: %s\nError: %s",
            $message->topic_name,
            $message->payload,
            $e->getMessage()
        ));
    }

    public function shutdown(): void
    {
        echo "Shutting down consumer...\n";
        $this->running = false;
    }

    public function __destruct()
    {
        $this->consumer->close();
    }
}

Consumer 코드도 꽤 길지? 하지만 여기에는 프로덕션 환경에서 필요한 많은 기능들이 들어있어:

• 자동 리밸런싱: 컨슈머가 추가되거나 제거될 때 파티션을 자동으로 재분배해
• 수동 커밋: 메시지 처리가 완료된 후에만 오프셋을 커밋해서 데이터 손실을 방지해
• 에러 처리: 메시지 처리 중 에러가 발생해도 안전하게 처리해
• 그레이스풀 셧다운: SIGTERM, SIGINT 시그널을 받으면 안전하게 종료해

📊 로그 처리 핸들러 구현

이제 실제로 로그를 처리하는 핸들러들을 만들어보자. 각 토픽별로 다른 처리 로직을 적용할 수 있어!

<?php
// src/LogProcessor.php

namespace App\Kafka;

use PDO;

class LogProcessor
{
    private PDO $db;
    private ElasticsearchClient $elasticsearch;

    public function __construct(PDO $db, ElasticsearchClient $elasticsearch)
    {
        $this->db = $db;
        $this->elasticsearch = $elasticsearch;
    }

    public function processUserAction(array $payload, $message): void
    {
        echo "Processing user action: {$payload['action']}\n";

        // 데이터베이스에 저장
        $stmt = $this->db->prepare("
            INSERT INTO user_actions 
            (user_id, action, data, timestamp, ip, user_agent)
            VALUES (?, ?, ?, ?, ?, ?)
        ");

        $stmt->execute([
            $payload['user_id'],
            $payload['action'],
            json_encode($payload['data']),
            date('Y-m-d H:i:s', $payload['timestamp']),
            $payload['ip'],
            $payload['user_agent'],
        ]);

        // Elasticsearch에도 인덱싱 (실시간 검색용)
        $this->elasticsearch->index('user-actions', [
            'user_id' => $payload['user_id'],
            'action' => $payload['action'],
            'data' => $payload['data'],
            'timestamp' => date('c', $payload['timestamp']),
            'application' => $payload['application'],
        ]);

        // 특정 액션에 대한 실시간 알림
        if ($payload['action'] === 'purchase') {
            $this->sendPurchaseNotification($payload);
        }
    }

    public function processApiLog(array $payload, $message): void
    {
        echo "Processing API log: {$payload['endpoint']}\n";

        // 느린 API 호출 감지
        if ($payload['duration'] > 1.0) {
            $this->alertSlowApi($payload);
        }

        // 에러 상태 코드 감지
        if ($payload['status_code'] >= 500) {
            $this->alertServerError($payload);
        }

        // 시계열 데이터베이스에 저장 (InfluxDB, TimescaleDB 등)
        $this->saveToTimeSeriesDB($payload);

        // 통계 업데이트
        $this->updateApiStats($payload);
    }

    public function processErrorLog(array $payload, $message): void
    {
        echo "Processing error log: {$payload['message']}\n";

        // 심각한 에러는 즉시 알림
        if ($payload['level'] === 'critical') {
            $this->sendCriticalAlert($payload);
        }

        // 에러 로그 저장
        $stmt = $this->db->prepare("
            INSERT INTO error_logs 
            (level, message, context, file, line, trace, timestamp, application)
            VALUES (?, ?, ?, ?, ?, ?, ?, ?)
        ");

        $stmt->execute([
            $payload['level'],
            $payload['message'],
            json_encode($payload['context']),
            $payload['file'],
            $payload['line'],
            $payload['trace'],
            date('Y-m-d H:i:s', $payload['timestamp']),
            $payload['application'],
        ]);

        // 에러 패턴 분석
        $this->analyzeErrorPattern($payload);
    }

    public function processPerformanceLog(array $payload, $message): void
    {
        echo "Processing performance log: {$payload['operation']}\n";

        // 성능 메트릭 저장
        $this->savePerformanceMetrics($payload);

        // 성능 저하 감지
        if ($payload['duration'] > $this->getThreshold($payload['operation'])) {
            $this->alertPerformanceDegradation($payload);
        }
    }

    private function sendPurchaseNotification(array $payload): void
    {
        // 구매 알림 전송 (이메일, 슬랙, SMS 등)
        echo "Sending purchase notification for user {$payload['user_id']}\n";
    }

    private function alertSlowApi(array $payload): void
    {
        // 느린 API 알림
        echo "ALERT: Slow API detected - {$payload['endpoint']} took {$payload['duration']}s\n";
    }

    private function alertServerError(array $payload): void
    {
        // 서버 에러 알림
        echo "ALERT: Server error - {$payload['endpoint']} returned {$payload['status_code']}\n";
    }

    private function sendCriticalAlert(array $payload): void
    {
        // 크리티컬 에러 알림
        echo "CRITICAL ALERT: {$payload['message']}\n";
    }

    private function saveToTimeSeriesDB(array $payload): void
    {
        // 시계열 DB에 저장
        echo "Saving to time series DB\n";
    }

    private function updateApiStats(array $payload): void
    {
        // API 통계 업데이트 (Redis 등 사용)
        echo "Updating API stats\n";
    }

    private function analyzeErrorPattern(array $payload): void
    {
        // 에러 패턴 분석
        echo "Analyzing error pattern\n";
    }

    private function savePerformanceMetrics(array $payload): void
    {
        // 성능 메트릭 저장
        echo "Saving performance metrics\n";
    }

    private function getThreshold(string $operation): float
    {
        // 작업별 임계값 반환
        $thresholds = [
            'database_query' => 0.5,
            'api_call' => 2.0,
            'file_upload' => 5.0,
        ];

        return $thresholds[$operation] ?? 1.0;
    }

    private function alertPerformanceDegradation(array $payload): void
    {
        // 성능 저하 알림
        echo "ALERT: Performance degradation - {$payload['operation']} took {$payload['duration']}s\n";
    }
}

🚀 Consumer 실행 스크립트

이제 모든 걸 조합해서 실제로 실행할 수 있는 스크립트를 만들어보자!

<?php
// bin/consume-logs.php

require_once __DIR__ . '/../vendor/autoload.php';

use App\Kafka\Consumer;
use App\Kafka\LogProcessor;

// 데이터베이스 연결
$db = new PDO(
    'mysql:host=localhost;dbname=logs',
    'username',
    'password',
    [PDO::ATTR_ERRMODE => PDO::ERRMODE_EXCEPTION]
);

// Elasticsearch 클라이언트 (가상)
$elasticsearch = new ElasticsearchClient([
    'hosts' => ['localhost:9200']
]);

// 로그 프로세서 생성
$processor = new LogProcessor($db, $elasticsearch);

// Consumer 생성
$consumer = new Consumer([
    'metadata.broker.list' => 'localhost:9092',
    'group.id' => 'log-processor-group',
    'auto.offset.reset' => 'earliest',
]);

// 토픽 구독
$consumer->subscribe([
    'user-actions',
    'api-logs',
    'error-logs',
    'critical-logs',
    'performance-logs',
]);

// 핸들러 등록
$consumer->registerHandler('user-actions', [$processor, 'processUserAction']);
$consumer->registerHandler('api-logs', [$processor, 'processApiLog']);
$consumer->registerHandler('error-logs', [$processor, 'processErrorLog']);
$consumer->registerHandler('critical-logs', [$processor, 'processErrorLog']);
$consumer->registerHandler('performance-logs', [$processor, 'processPerformanceLog']);

// 소비 시작!
echo "Starting log consumer...\n";
$consumer->consume();

이제 php bin/consume-logs.php 명령어로 실행하면 Consumer가 계속 실행되면서 Kafka에서 메시지를 가져와 처리해! 🎉

대규모 시스템에서 가장 중요한 건 안정성이야. 시스템이 다운되면 로그가 손실될 수 있고, 이는 곧 중요한 정보를 놓치는 거니까! 😰

🔄 재시도 및 에러 처리

<?php
// src/RetryHandler.php

namespace App\Kafka;

class RetryHandler
{
    private int $maxRetries;
    private int $retryDelay;
    private KafkaProducer $dlqProducer;

    public function __construct(
        int $maxRetries = 3,
        int $retryDelay = 1000,
        ?KafkaProducer $dlqProducer = null
    ) {
        $this->maxRetries = $maxRetries;
        $this->retryDelay = $retryDelay;
        $this->dlqProducer = $dlqProducer;
    }

    public function handle(callable $callback, array $payload, Message $message): bool
    {
        $attempts = 0;
        $lastException = null;

        while ($attempts < $this->maxRetries) {
            try {
                $callback($payload, $message);
                return true;
            } catch (\Exception $e) {
                $attempts++;
                $lastException = $e;

                echo sprintf(
                    "Retry %d/%d failed: %s\n",
                    $attempts,
                    $this->maxRetries,
                    $e->getMessage()
                );

                if ($attempts < $this->maxRetries) {
                    usleep($this->retryDelay * 1000 * $attempts);  // 지수 백오프
                }
            }
        }

        // 모든 재시도 실패 시 DLQ로 전송
        if ($this->dlqProducer) {
            $this->sendToDeadLetterQueue($payload, $message, $lastException);
        }

        return false;
    }

    private function sendToDeadLetterQueue(
        array $payload,
        Message $message,
        \Exception $exception
    ): void {
        $dlqPayload = [
            'original_topic' => $message->topic_name,
            'original_partition' => $message->partition,
            'original_offset' => $message->offset,
            'payload' => $payload,
            'error' => [
                'message' => $exception->getMessage(),
                'trace' => $exception->getTraceAsString(),
            ],
            'timestamp' => time(),
        ];

        $this->dlqProducer->send(
            'dead-letter-queue',
            json_encode($dlqPayload)
        );

        echo "Message sent to Dead Letter Queue\n";
    }
}

📈 모니터링 시스템 구축

시스템이 잘 돌아가는지 확인하려면 모니터링이 필수야! Prometheus와 Grafana를 사용하면 멋진 대시보드를 만들 수 있어. 📊

<?php
// src/Metrics.php

namespace App\Kafka;

class Metrics
{
    private array $counters = [];
    private array $gauges = [];
    private array $histograms = [];

    public function incrementCounter(string $name, int $value = 1, array $labels = []): void
    {
        $key = $this->makeKey($name, $labels);
        
        if (!isset($this->counters[$key])) {
            $this->counters[$key] = 0;
        }
        
        $this->counters[$key] += $value;
    }

    public function setGauge(string $name, float $value, array $labels = []): void
    {
        $key = $this->makeKey($name, $labels);
        $this->gauges[$key] = $value;
    }

    public function observeHistogram(string $name, float $value, array $labels = []): void
    {
        $key = $this->makeKey($name, $labels);
        
        if (!isset($this->histograms[$key])) {
            $this->histograms[$key] = [];
        }
        
        $this->histograms[$key][] = $value;
    }

    public function export(): string
    {
        $output = [];

        // Counters
        foreach ($this->counters as $key => $value) {
            $output[] = "{$key} {$value}";
        }

        // Gauges
        foreach ($this->gauges as $key => $value) {
            $output[] = "{$key} {$value}";
        }

        // Histograms
        foreach ($this->histograms as $key => $values) {
            $count = count($values);
            $sum = array_sum($values);
            $output[] = "{$key}_count {$count}";
            $output[] = "{$key}_sum {$sum}";
        }

        return implode("\n", $output);
    }

    private function makeKey(string $name, array $labels): string
    {
        if (empty($labels)) {
            return $name;
        }

        $labelStr = [];
        foreach ($labels as $key => $value) {
            $labelStr[] = "{$key}=\"{$value}\"";
        }

        return $name . '{' . implode(',', $labelStr) . '}';
    }
}

// 사용 예시
$metrics = new Metrics();

// 메시지 처리 카운터
$metrics->incrementCounter('kafka_messages_processed_total', 1, [
    'topic' => 'user-actions',
    'status' => 'success',
]);

// Consumer lag 게이지
$metrics->setGauge('kafka_consumer_lag', 1234, [
    'topic' => 'user-actions',
    'partition' => '0',
]);

// 처리 시간 히스토그램
$metrics->observeHistogram('kafka_message_processing_duration_seconds', 0.125, [
    'topic' => 'user-actions',
]);

🚨 알림 시스템

문제가 발생하면 즉시 알아야 해! 슬랙, 이메일, SMS 등으로 알림을 보내는 시스템을 만들어보자.

<?php
// src/AlertManager.php

namespace App\Kafka;

class AlertManager
{
    private array $channels = [];
    private array $alertHistory = [];
    private int $cooldownPeriod = 300;  // 5분

    public function addChannel(string $name, callable $sender): void
    {
        $this->channels[$name] = $sender;
    }

    public function sendAlert(
        string $level,
        string $message,
        array $context = [],
        array $channels = []
    ): void {
        // 중복 알림 방지
        $alertKey = md5($level . $message);
        $now = time();

        if (isset($this->alertHistory[$alertKey])) {
            $lastSent = $this->alertHistory[$alertKey];
            if ($now - $lastSent < $this->cooldownPeriod) {
                return;  // 쿨다운 기간 내에는 같은 알림 보내지 않음
            }
        }

        $this->alertHistory[$alertKey] = $now;

        // 알림 전송
        $alertData = [
            'level' => $level,
            'message' => $message,
            'context' => $context,
            'timestamp' => date('Y-m-d H:i:s'),
            'hostname' => gethostname(),
        ];

        $channelsToUse = empty($channels) ? array_keys($this->channels) : $channels;

        foreach ($channelsToUse as $channelName) {
            if (isset($this->channels[$channelName])) {
                try {
                    $this->channels[$channelName]($alertData);
                } catch (\Exception $e) {
                    error_log("Failed to send alert via {$channelName}: " . $e->getMessage());
                }
            }
        }
    }

    public function critical(string $message, array $context = []): void
    {
        $this->sendAlert('CRITICAL', $message, $context);
    }

    public function warning(string $message, array $context = []): void
    {
        $this->sendAlert('WARNING', $message, $context);
    }

    public function info(string $message, array $context = []): void
    {
        $this->sendAlert('INFO', $message, $context);
    }
}

// 사용 예시
$alertManager = new AlertManager();

// 슬랙 채널 추가
$alertManager->addChannel('slack', function($alert) {
    $webhook = 'https://hooks.slack.com/services/YOUR/WEBHOOK/URL';
    
    $payload = [
        'text' => "[{$alert['level']}] {$alert['message']}",
        'attachments' => [
            [
                'color' => $alert['level'] === 'CRITICAL' ? 'danger' : 'warning',
                'fields' => [
                    [
                        'title' => 'Hostname',
                        'value' => $alert['hostname'],
                        'short' => true,
                    ],
                    [
                        'title' => 'Timestamp',
                        'value' => $alert['timestamp'],
                        'short' => true,
                    ],
                ],
            ],
        ],
    ];

    $ch = curl_init($webhook);
    curl_setopt($ch, CURLOPT_POST, 1);
    curl_setopt($ch, CURLOPT_POSTFIELDS, json_encode($payload));
    curl_setopt($ch, CURLOPT_HTTPHEADER, ['Content-Type: application/json']);
    curl_setopt($ch, CURLOPT_RETURNTRANSFER, true);
    curl_exec($ch);
    curl_close($ch);
});

// 이메일 채널 추가
$alertManager->addChannel('email', function($alert) {
    mail(
        'admin@example.com',
        "[{$alert['level']}] Kafka Alert",
        print_r($alert, true)
    );
});

// 알림 전송
$alertManager->critical('Consumer lag exceeded threshold', [
    'topic' => 'user-actions',
    'lag' => 10000,
    'threshold' => 5000,
]);

기본 시스템을 만들었으니 이제 더 고급 기능들을 추가해보자! 실제 프로덕션 환경에서는 이런 기능들이 정말 유용해. 🎪

🔄 스트림 처리 (Kafka Streams)

Kafka Streams를 사용하면 실시간으로 데이터를 변환하고 집계할 수 있어. PHP에서는 직접 Kafka Streams API를 사용할 수 없지만, Consumer에서 비슷한 기능을 구현할 수 있어!

<?php
// src/StreamProcessor.php

namespace App\Kafka;

class StreamProcessor
{
    private array $windows = [];
    private int $windowSize = 60;  // 60초 윈도우

    public function processWithWindow(array $payload): void
    {
        $timestamp = $payload['timestamp'];
        $windowKey = floor($timestamp / $this->windowSize) * $this->windowSize;

        if (!isset($this->windows[$windowKey])) {
            $this->windows[$windowKey] = [
                'count' => 0,
                'sum' => 0,
                'min' => PHP_FLOAT_MAX,
                'max' => PHP_FLOAT_MIN,
                'events' => [],
            ];
        }

        $window = &$this->windows[$windowKey];
        $window['count']++;
        
        if (isset($payload['value'])) {
            $value = $payload['value'];
            $window['sum'] += $value;
            $window['min'] = min($window['min'], $value);
            $window['max'] = max($window['max'], $value);
        }

        $window['events'][] = $payload;

        // 오래된 윈도우 정리
        $this->cleanOldWindows($timestamp);

        // 윈도우가 완료되면 집계 결과 저장
        if ($this->isWindowComplete($windowKey, $timestamp)) {
            $this->finalizeWindow($windowKey);
        }
    }

    private function cleanOldWindows(float $currentTimestamp): void
    {
        $cutoff = floor(($currentTimestamp - $this->windowSize * 10) / $this->windowSize) * $this->windowSize;

        foreach ($this->windows as $key => $window) {
            if ($key < $cutoff) {
                unset($this->windows[$key]);
            }
        }
    }

    private function isWindowComplete(int $windowKey, float $currentTimestamp): bool
    {
        return $currentTimestamp > ($windowKey + $this->windowSize);
    }

    private function finalizeWindow(int $windowKey): void
    {
        $window = $this->windows[$windowKey];

        $aggregated = [
            'window_start' => date('Y-m-d H:i:s', $windowKey),
            'window_end' => date('Y-m-d H:i:s', $windowKey + $this->windowSize),
            'count' => $window['count'],
            'average' => $window['count'] > 0 ? $window['sum'] / $window['count'] : 0,
            'min' => $window['min'] !== PHP_FLOAT_MAX ? $window['min'] : null,
            'max' => $window['max'] !== PHP_FLOAT_MIN ? $window['max'] : null,
        ];

        // 집계 결과 저장
        $this->saveAggregatedData($aggregated);

        echo "Window finalized: " . json_encode($aggregated) . "\n";
    }

    private function saveAggregatedData(array $data): void
    {
        // 데이터베이스나 다른 Kafka 토픽으로 저장
        echo "Saving aggregated data\n";
    }
}

🎭 데이터 변환 파이프라인

로그 데이터를 다양한 형태로 변환해서 여러 목적으로 사용할 수 있어. ETL 파이프라인을 만들어보자!

<?php
// src/Pipeline/TransformPipeline.php

namespace App\Pipeline;

class TransformPipeline
{
    private array $transformers = [];

    public function addTransformer(callable $transformer): self
    {
        $this->transformers[] = $transformer;
        return $this;
    }

    public function process(array $data): array
    {
        $result = $data;

        foreach ($this->transformers as $transformer) {
            $result = $transformer($result);
            
            if ($result === null) {
                return null;  // 필터링됨
            }
        }

        return $result;
    }
}

// 사용 예시
$pipeline = new TransformPipeline();

// 1. 데이터 정제
$pipeline->addTransformer(function($data) {
    // null 값 제거
    return array_filter($data, function($value) {
        return $value !== null;
    });
});

// 2. 타입 변환
$pipeline->addTransformer(function($data) {
    if (isset($data['timestamp']) && is_string($data['timestamp'])) {
        $data['timestamp'] = strtotime($data['timestamp']);
    }
    return $data;
});

// 3. 필드 추가
$pipeline->addTransformer(function($data) {
    $data['processed_at'] = time();
    $data['version'] = '1.0';
    return $data;
});

// 4. 민감 정보 마스킹
$pipeline->addTransformer(function($data) use ($masker) {
    return $masker->mask($data);
});

// 5. 필터링 (특정 조건에 맞는 것만)
$pipeline->addTransformer(function($data) {
    if (isset($data['level']) && $data['level'] === 'debug') {
        return null;  // debug 로그는 제외
    }
    return $data;
});

// 파이프라인 실행
$transformed = $pipeline->process($rawData);

📈 실시간 분석 및 알림

로그를 실시간으로 분석해서 이상 징후를 감지하고 즉시 알림을 보낼 수 있어!

<?php
// src/Analytics/AnomalyDetector.php

namespace App\Analytics;

class AnomalyDetector
{
    private array $baselines = [];
    private float $threshold = 2.0;  // 표준편차의 2배

    public function detect(string $metric, float $value): bool
    {
        if (!isset($this->baselines[$metric])) {
            $this->baselines[$metric] = [
                'values' => [],
                'mean' => 0,
                'stddev' => 0,
            ];
        }

        $baseline = &$this->baselines[$metric];
        $baseline['values'][] = $value;

        // 최근 100개 값만 유지
        if (count($baseline['values']) > 100) {
            array_shift($baseline['values']);
        }

        // 평균과 표준편차 계산
        $baseline['mean'] = array_sum($baseline['values']) / count($baseline['values']);
        $baseline['stddev'] = $this->calculateStdDev($baseline['values'], $baseline['mean']);

        // 이상 징후 감지
        $deviation = abs($value - $baseline['mean']);
        $isAnomaly = $deviation > ($baseline['stddev'] * $this->threshold);

        if ($isAnomaly) {
            echo "ANOMALY DETECTED: {$metric} = {$value} (mean: {$baseline['mean']}, stddev: {$baseline['stddev']})\n";
        }

        return $isAnomaly;
    }

    private function calculateStdDev(array $values, float $mean): float
    {
        $variance = 0;
        
        foreach ($values as $value) {
            $variance += pow($value - $mean, 2);
        }

        $variance /= count($values);

        return sqrt($variance);
    }
}

// 사용 예시
$detector = new AnomalyDetector();

// API 응답 시간 모니터링
foreach ($apiLogs as $log) {
    $isAnomaly = $detector->detect('api_response_time', $log['duration']);
    
    if ($isAnomaly) {
        $alertManager->warning('Abnormal API response time detected', [
            'endpoint' => $log['endpoint'],
            'duration' => $log['duration'],
        ]);
    }
}

// 에러율 모니터링
$errorRate = $errorCount / $totalRequests;
$isAnomaly = $detector->detect('error_rate', $errorRate);

if ($isAnomaly) {
    $alertManager->critical('Error rate spike detected', [
        'error_rate' => $errorRate,
        'error_count' => $errorCount,
        'total_requests' => $totalRequests,
    ]);
}