Distributed Tracing với Zipkin + Kafka + Docker + Spring Boot: Tracing - Hành trình trở thành Sherlock Dev

version: '2'
services:
  zookeeper-1:
    image: confluentinc/cp-zookeeper:latest
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181
      ZOOKEEPER_TICK_TIME: 2000
    ports:
      - 22181:2181

  kafka-1:
    container_name: kafka1
    image: confluentinc/cp-kafka:latest
    depends_on:
      - zookeeper-1

    ports:
      - 29092:29092
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: zookeeper-1:2181
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: EXTERNAL:PLAINTEXT,INTERNAL:PLAINTEXT
      KAFKA_ADVERTISED_LISTENERS: EXTERNAL://localhost:29092,INTERNAL://kafka1:9092
      KAFKA_INTER_BROKER_LISTENER_NAME: INTERNAL
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
  kafka-2:
    container_name: kafka2
    image: confluentinc/cp-kafka:latest
    depends_on:
      - zookeeper-1
    ports:
      - 29093:29093
    environment:
      KAFKA_BROKER_ID: 2
      KAFKA_ZOOKEEPER_CONNECT: zookeeper-1:2181
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: EXTERNAL:PLAINTEXT,INTERNAL:PLAINTEXT
      KAFKA_ADVERTISED_LISTENERS: EXTERNAL://localhost:29093,INTERNAL://kafka2:9092
      KAFKA_INTER_BROKER_LISTENER_NAME: INTERNAL
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
  kafdrop:
    image: obsidiandynamics/kafdrop
    restart: "no"
    environment:
      KAFKA_BROKERCONNECT: "kafka2:9092,kafka1:9092"
    ports:
      - "9002:9000"
    depends_on:
      - kafka-1
      - kafka-2
  zipkin:
    image: openzipkin/zipkin
    container_name: zipkin
    environment:
      - STORAGE_TYPE=mysql
      # Point the zipkin at the storage backend
      - MYSQL_HOST=host.docker.internal
      - MYSQL_USER=review
      - MYSQL_PASS=Techlens@321
      - MYSQL_DB=zipkin
      - MYSQL_TCP_PORT=3306
      - MYSQL_MAX_CONNECTIONS=10
      # Uncomment to enable scribe
      # - SCRIBE_ENABLED=true
      # Uncomment to enable self-tracing
      # - SELF_TRACING_ENABLED=true
      # Uncomment to enable debug logging
      - JAVA_OPTS=-Dlogging.level.zipkin2=DEBUG -Dlogging.level.org.apache.kafka=DEBUG
      - KAFKA_BOOTSTRAP_SERVERS=kafka-1:9092
      - KAFKA_TOPIC=zipkin
    ports:
      # Port used for the Zipkin UI and HTTP Api
      - 9411:9411
    depends_on:
      - kafka-1
      - kafka-2

docker-compose -f .\kafka-compose.yml  up -d

docker-compose -f .\kafka-compose.yml ps

docker-compose -f .\kafka-compose.yml exec kafka-1 kafka-topics.sh --create --topic zipkin --consumer-property group.id=zipkin --partitions 2 --replication-factor 2 --bootstrap-server kafka-1:9092

<!-- https://mvnrepository.com/artifact/io.micrometer/micrometer-tracing -->
<dependency>
  <groupId>io.micrometer</groupId>
  <artifactId>micrometer-tracing</artifactId>
  <version>1.2.0</version>
</dependency>
<!-- https://mvnrepository.com/artifact/org.springframework.kafka/spring-kafka -->
<dependency>
  <groupId>org.springframework.kafka</groupId>
  <artifactId>spring-kafka</artifactId>
  <version>3.1.0</version>
  </dependency>
<!-- https://mvnrepository.com/artifact/io.zipkin.reporter2/zipkin-sender-kafka -->
<dependency>
  <groupId>io.zipkin.reporter2</groupId>
  <artifactId>zipkin-sender-kafka</artifactId>
  <version>2.16.4</version>
</dependency>

spring:
  ...

  kafka:
    bootstrap-servers: localhost:29092
    consumer:
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      group-id: zipkin
    producer:
      acks: 0 # gửi không an toàn, gửi không nhận ack
      batch-size: 4
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.apache.kafka.common.serialization.StringSerializer

  zipkin:
    kafka:
      topic: zipkin
      group-id: zipkin

logging:
  pattern:
    level: '%5p [${spring.application.name:},%X{traceId:-},%X{spanId:-}]'

management:
  tracing:
    sampling:
      probability: 1.0

@Configuration
@EnableConfigurationProperties(KafkaProperties.class)
public class KafkaConfig {

    static String join(List<?> parts) {
        StringBuilder to = new StringBuilder();
        for (int i = 0, length = parts.size(); i < length; i++) {
            to.append(parts.get(i));
            if (i + 1 < length) {
                to.append(',');
            }
        }
        return to.toString();
    }

    @Bean("zipkinSender")
    Sender kafkaSender(KafkaProperties config, Environment environment) {
        // Need to get property value from Environment
        // because when using @VaultPropertySource in reactive web app
        // this bean is initiated before @Value is resolved
        // See gh-1990
        String topic = environment.getProperty("spring.zipkin.kafka.topic", "zipkin");
        String groupId = environment.getProperty("spring.zipkin.kafka.group-id", "zipkin");
        Map<String, Object> properties = new HashMap<>();
        properties.put("key.serializer", ByteArraySerializer.class);
        properties.put("value.serializer", ByteArraySerializer.class);
        properties.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
        // Kafka expects the input to be a String, but KafkaProperties returns a list
        List<String> bootstrapServers = config.getBootstrapServers();
            properties.put("bootstrap.servers", join(bootstrapServers));
        return KafkaSender.newBuilder().topic(topic).overrides(properties).build();
    }
}

@RestController
public class PingController {

    Logger logger = LoggerFactory.getLogger(PingController.class);
    @GetMapping("/ping")
    public String ping() {
        logger.info("pong");
        return "pong";
    }
}

curl --location 'localhost:6000/techlens/ping'