<dependency>
    <groupId>com.luamas.thinkbridge</groupId>
    <artifactId>web-spring-boot-starter</artifactId>
</dependency>

implementation("com.luamas.thinkbridge:web-spring-boot-starter")

<dependency>
    <groupId>com.luamas.thinkbridge</groupId>
    <artifactId>webflux-spring-boot-starter</artifactId>
</dependency>

implementation("com.luamas.thinkbridge:webflux-spring-boot-starter")

think:
  log:
    receive:
      standard:
        kafka:
          # 是否启用，默认启用
          enabled: true
          # 是否异步发送，默认true
          async: true
          trace:
            # 主题
            topic: thinkBridge-log
            # 指定分区，默认不指定
            partition:
            # Kafka Producer配置，参考org.apache.kafka.clients.producer.ProducerConfig
            properties:
              bootstrap.servers: localhost:9092
              key.serializer: org.apache.kafka.common.serialization.ByteArraySerializer
              value.serializer: org.apache.kafka.common.serialization.ByteArraySerializer

think:
  log:
    receive:
      standard:
        amqp:
          # 是否启用，默认启用
          enabled: true
          # 是否异步发送，默认true
          async: true
          trace:
            # 主机，默认localhost
            host: localhost
            # 端口，默认5672
            port: 5672
            # 用户名，默认guest
            username: guest
            # 密码，默认guest
            password: guest
            # 虚拟主机，默认/
            virtual-host: /
            # 交换机，默认thinkbridge.log（必填）
            exchange: thinkbridge.log
            # 交换机类型：topic,direct,fanout,headers，默认topic
            exchange-type: topic
            # 路由key，默认thinkbridge.log.trace.common
            routing-key: thinkbridge.log.trace.common
            # 消费者队列，默认thinkbridge.log.logstash
            consumer-queue: thinkbridge.log.logstash
            # 给队列绑定的路由key，默认thinkbridge.log.trace.*
            consumer-queue-routing-key: thinkbridge.log.trace.*
            # 是否初始化exchange，默认false
            init-exchange: true
            # 是否初始化消费者queue，默认false
            init-consumer-queue-declare: true
            # 是否初始化绑定关系，默认false
            init-consumer-queue-bind: true
            # 是否持久化，默认true
            durable: true
            # 是否排他，默认false
            exclusive: false
            # 是否自动删除，默认false
            auto-delete: false

think:
  log:
    receive:
      standard:
        pulsar:
          # 是否启用，默认启用
          enabled: true
          # 是否异步发送，默认true
          async: true
          trace:
            # 主题
            topic: thinkBridge-log
            # Pulsar服务地址
            service-url: pulsar://localhost:6650
            # Pulsar Producer配置
            properties:
              # 可自定义 Pulsar Producer 属性，如：
              # send-timeout-ms: 30000
              # max-pending-messages: 1000

@Service
public class ManualLogService {
    @Autowired
    private EventService eventService;
}

public void sendOperationLog() {
    OperationLog operationLog = new OperationLog();
    operationLog.setMsg("手动发送的操作日志");
    operationLog.setClassName(this.getClass().getName());
    operationLog.setMethodName("sendOperationLog");
    operationLog.setSuccess(true);
    operationLog.setStartTime(new Date());
    operationLog.setTime(0L);

    // 日志类型默认为 OperationLog（模型类名）
    eventService.publishEvent(new OperationEvent(operationLog));

    // 指定自定义日志类型
    eventService.publishEvent(new OperationEvent(operationLog), "MyOperationLog");
}

public void sendAccessLog() {
    AccessLog accessLog = new AccessLog();
    accessLog.setUri("/manual/test");
    accessLog.setMethod("GET");
    accessLog.setSuccess(true);
    accessLog.setStartTime(new Date());
    accessLog.setTime(10L);

    eventService.publishEvent(new AccessEvent(accessLog));
}

public void sendErrorLog() {
    try {
        // 业务逻辑
    } catch (Exception e) {
        OperationLog operationLog = new OperationLog();
        operationLog.setMsg("执行失败");
        operationLog.setSuccess(false);
        operationLog.setStartTime(new Date());
        operationLog.setTime(0L);
        eventService.publishEvent(new OperationEvent(operationLog, e));
    }
}

public void sendLogWithExtend() {
    OperationLog operationLog = new OperationLog();
    operationLog.setMsg("手动发送的操作日志");
    operationLog.setSuccess(true);
    operationLog.setStartTime(new Date());
    operationLog.setTime(0L);

    // 存储自定义业务字段
    operationLog.getExtend().put("userId", "1001");
    operationLog.getExtend().put("tenantId", "thinkbridge");
    operationLog.getExtend().put("amount", 99.9);

    eventService.publishEvent(new OperationEvent(operationLog));
}

{
  "msg": "手动发送的操作日志",
  "success": true,
  "extend": {
    "userId": "1001",
    "tenantId": "thinkbridge",
    "amount": 99.9
  }
}

Map<String, Object> extend = new HashMap<>();
extend.put("userId", "1001");
extend.put("tenantId", "thinkbridge");
operationLog.setExtend(extend);

@Getter
@Setter
public class CustomLog extends AbstractMessageLog {
    private String content;
}

public class CustomEvent extends LogEvent<CustomLog> {
    public CustomEvent(CustomLog source) {
        super(source);
    }

    public CustomEvent(CustomLog source, Throwable throwable) {
        super(source, throwable);
    }
}

public void sendCustomLog() {
    CustomLog customLog = new CustomLog();
    customLog.setContent("自定义日志内容");
    customLog.setSuccess(true);
    customLog.setStartTime(new Date());
    customLog.setTime(0L);
    eventService.publishEvent(new CustomEvent(customLog));
}

think:
  log:
    receive:
      logback:
        kafka:
          # 是否启用，默认启用
          enabled: true
          # 是否异步发送，默认true
          async: true
          trace:
            # Kafka服务器地址
            bootstrap: localhost:9092
            # 主题
            topic: thinkbridge-kafka-trace-log
            # ack配置，默认1
            ack: 1
            # 重试次数，默认3
            retries: 3
            # 发送大小限制，默认10MB
            size: 10485760

<?xml version="1.0" encoding="UTF-8"?>
<configuration scan="true" scanPeriod="1 seconds">
    <contextName>logback</contextName>
    <contextListener class="ch.qos.logback.classic.jul.LevelChangePropagator">
        <resetJUL>true</resetJUL>
    </contextListener>
	<!-- 这里一定要引入此配置，否则无法使用 -->
    <include resource="thinkbridge/logback-kafka-trace-log.xml" />

    <!-- 控制台输出 -->
    <appender name="CONSOLE" class="ch.qos.logback.core.ConsoleAppender">
        <encoder class="ch.qos.logback.classic.encoder.PatternLayoutEncoder">
            <!--格式化输出：%d表示日期，%thread表示线程名，%-5level：级别从左显示5个字符宽度%msg：日志消息，%n是换行符-->
            <pattern>%d{yyyy-MM-dd HH:mm:ss} %highlight(%-5p) [%thread] %logger.%M\(%F:%L\) :[%X\{logType}]
                %msg%n
            </pattern>
        </encoder>
    </appender>
    <!-- 关闭apache包信息-->
    <logger name="org.apache.catalina.core" level="OFF">
        <appender-ref ref="CONSOLE"/>
    </logger>
    <!-- 默认级别 -->
    <root level="INFO">
        <appender-ref ref="CONSOLE"/>
    </root>
</configuration>

think:
  logging:
    kafka:
      enabled: true
      bootstrap-servers: localhost:9092
      topic: app-logs
      app-name: my-app
      environment: default

      acks: "1"
      retries: 3
      batch-size: 16384
      linger-ms: 5
      buffer-memory: 33554432
      compression-type: none
      idempotence: false
      max-in-flight-requests: 5

      max-retries: 3
      retry-backoff-ms: 100
      max-retry-backoff-ms: 5000

      queue-size: 10000
      max-failed-logs-per-topic: 1000
      failed-logs-ttl-hours: 24

      include-mdc: true
      include-caller-data: false
      exclude-fields:
        - password
        - token
        - secret
        - key
      exclude-loggers:
        - org.apache.kafka
        - org.springframework
      min-level: INFO

<?xml version="1.0" encoding="UTF-8"?>
<configuration scan="true" scanPeriod="1 seconds">
    <contextName>logback</contextName>

    <appender name="CONSOLE" class="ch.qos.logback.core.ConsoleAppender">
        <encoder class="ch.qos.logback.classic.encoder.PatternLayoutEncoder">
            <pattern>%d{yyyy-MM-dd HH:mm:ss} %highlight(%-5p) [%thread] %logger.%M\(%F:%L\) :[%X\{logType}] %msg%n</pattern>
        </encoder>
    </appender>

    <root level="INFO">
        <appender-ref ref="CONSOLE"/>
    </root>
</configuration>

think:
  logging:
    pulsar:
      enabled: true
      service-url: pulsar://localhost:6650
      topic: app-logs
      app-name: my-application
      environment: production
      send-timeout-ms: 30000
      max-pending-messages: 1000
      block-if-queue-full: false
      enable-batching: true
      batching-max-messages: 1000
      batching-max-publish-delay-ms: 10
      batching-max-bytes: 131072
      max-retries: 3
      retry-backoff-ms: 100
      max-retry-backoff-ms: 5000
      queue-size: 10000
      max-failed-logs-per-topic: 1000
      failed-logs-ttl-hours: 24
      include-mdc: true
      include-caller-data: false
      exclude-fields:
        - password
        - token
        - secret
        - key
      exclude-loggers:
        - org.apache.pulsar
        - org.springframework
      min-level: INFO

<?xml version="1.0" encoding="UTF-8"?>
<configuration>
    <include resource="org/springframework/boot/logging/logback/base.xml"/>

    <appender name="PULSAR" class="com.luamas.thinkbridge.bootstarter.log.logback.pulsar.appender.PulsarAppender">
        <enabled>true</enabled>
        <topic>app-logs</topic>
        <serviceUrl>pulsar://localhost:6650</serviceUrl>
        <appName>my-application</appName>
        <environment>production</environment>
        <includeMdc>true</includeMdc>
        <includeCallerData>false</includeCallerData>
        <maxRetries>3</maxRetries>
        <queueSize>10000</queueSize>
    </appender>

    <root level="INFO">
        <appender-ref ref="PULSAR"/>
    </root>
</configuration>

{
  "@timestamp": 1692000000000,
  "timestamp": "2023-08-14T08:00:00.000Z",
  "level": "INFO",
  "logger": "com.example.MyService",
  "message": "User login successful",
  "thread": "http-nio-8080-exec-1",
  "app": "my-application",
  "env": "production",
  "host": "my-server",
  "ip": "192.168.1.100",
  "mdc": {
    "userId": "12345",
    "traceId": "abc-def-ghi"
  },
  "exception": null,
  "extra": {
    "pid": "1234",
    "logTime": "2023-08-14T08:00:00.123Z"
  }
}

延迟 = min(baseDelay * 2^(attempt-1), maxDelay) * jitter(0.8~1.2)
重试 1: ~100ms
重试 2: ~200ms
重试 3: ~400ms

think:
  log:
    receive:
      logback:
        amqp:
          # 是否启用，默认启用
          enabled: true
          # 是否异步发送，默认true
          async: true
          trace:
            # RabbitMQ主机
            host: 127.0.0.1
            # RabbitMQ端口
            port: 5672
            # 用户名
            username: guest
            # 密码
            password: guest
            # 虚拟主机
            virtual-host: /
            # 交换机名称
            exchange-name: thinkBridge-log
            # 交换机类型，默认topic
            exchange-type: topic
            # 交换机不存在时自动创建
            declare-exchange: true
            # 是否持久化
            durable: true
            # 是否自动删除
            auto-delete: false

<?xml version="1.0" encoding="UTF-8"?>
<configuration scan="true" scanPeriod="1 seconds">
    <contextName>logback</contextName>
    <contextListener class="ch.qos.logback.classic.jul.LevelChangePropagator">
        <resetJUL>true</resetJUL>
    </contextListener>
	<!-- 这里一定要引入此配置，否则无法使用 -->
    <include resource="thinkbridge/logback-amqp-trace-log.xml" />

    <!-- 控制台输出 -->
    <appender name="CONSOLE" class="ch.qos.logback.core.ConsoleAppender">
        <encoder class="ch.qos.logback.classic.encoder.PatternLayoutEncoder">
            <!--格式化输出：%d表示日期，%thread表示线程名，%-5level：级别从左显示5个字符宽度%msg：日志消息，%n是换行符-->
            <pattern>%d{yyyy-MM-dd HH:mm:ss} %highlight(%-5p) [%thread] %logger.%M\(%F:%L\) :[%X\{logType}]
                %msg%n
            </pattern>
        </encoder>
    </appender>
    <!-- 关闭apache包信息-->
    <logger name="org.apache.catalina.core" level="OFF">
        <appender-ref ref="CONSOLE"/>
    </logger>
    <!-- 默认级别 -->
    <root level="INFO">
        <appender-ref ref="CONSOLE"/>
    </root>
</configuration>

input {
  kafka {
    bootstrap_servers => "<kafka-bootstrap-servers>"
    topics => ["<kafka-topic>"] # 订阅主题
    auto_offset_reset => "earliest" #从最早的偏移量开始消费
    group_id => "<consumer-group-id>" # 组id
    consumer_threads => 5
  }
}

filter {
  # 解析kafka message
  json {
    source => "message"
    remove_field => ["event","@version"]
  }

  mutate {
      add_field => { "messagetmp" => "%{message}" }
  }
  json {
      source => "messagetmp"
      remove_field => ["messagetmp", "message"]
  }

  mutate {
    rename => { "logType" => "log_type" }
  }

  mutate {
    add_field => { "[@metadata][log_type]" => "%{log_type}" }
    add_field => { "[@metadata][application_name]" => "%{application_name}" }
  }

  mutate {
    lowercase => ["[@metadata][log_type]", "[@metadata][application_name]"]
  }
  mutate {
    remove_field => ["@timestamp"]
    rename => { "start_time" => "@timestamp" }
  }
}

output {
  elasticsearch {
    hosts => ["<elasticsearch-host>"]
    index => "thinkbridge-trace-%{[@metadata][application_name]}-%{[@metadata][log_type]}"
    user => "<elasticsearch-username>" #用户名
    password => "<elasticsearch-password>" #密码
    # for data stream
    # action => "create" #这里如果使用es的 stream滚动收集需要打开此注释
  }
}

input {
  rabbitmq{
      host=>"<rabbitmq-host>"                # 这里填写Rabbitmq的地址，确保可以ping通
      port=> 15672                            # 这里填写Rabbitmq的端口
      user=>"<rabbitmq-username>"                          # 这里填写Rabbitmq的用户名
      password=>"<rabbitmq-password>"                      # 这里填写Rabbitmq的密码
      exchange => "<rabbitmq-exchange>"      # 这里填写Rabbitmq的交换机的名称
      key => "<routing-key>"                   #路由键
      queue=>"<rabbitmq-queue>"                     # 这里填写Rabbitmq的队列的名称
      durable=> true                         # 这里填写Rabbitmq的队列的durable属性
      codec=>json                            # 这里填写Rabbitmq的队列的内容是什么格式
     }
}

filter {
  # 解析rabbitmq message
  json {
    source => "message"
    remove_field => ["event","@version"]
  }

  mutate {
      add_field => { "messagetmp" => "%{message}" }
  }
  json {
      source => "messagetmp"
      remove_field => ["messagetmp", "message"]
  }

  mutate {
    rename => { "logType" => "log_type" }
  }

  mutate {
    add_field => { "[@metadata][log_type]" => "%{log_type}" }
    add_field => { "[@metadata][application_name]" => "%{application_name}" }
  }

  mutate {
    lowercase => ["[@metadata][log_type]", "[@metadata][application_name]"]
  }
  mutate {
    remove_field => ["@timestamp"]
    rename => { "start_time" => "@timestamp" }
  }
}

output {
  elasticsearch {
    hosts => ["<elasticsearch-host>"]
    index => "thinkbridge-trace-%{[@metadata][application_name]}-%{[@metadata][log_type]}"
    user => "<elasticsearch-username>"
    password => "<elasticsearch-password>"
    # for data stream
    # action => "create"
  }
}

entry-pipeline:
  source:
    kafka:
      bootstrap_servers: "<kafka-bootstrap-servers>" # Kafka服务器地址
      topics:
        - name: "<kafka-topic>" # 订阅主题
      group_id: "<consumer-group-id>" # 组id
      auto_offset_reset: earliest # 从最早的偏移量开始消费
  processor:
    - parse_json:
        source: message # 解析kafka原始消息
    - delete_entries:
        with_keys:
          - event
          - "@version"
    - parse_json:
        source: message # 二次解析message字段
    - delete_entries:
        with_keys:
          - message
    - rename_keys:
        entries:
          - from_key: logType # 重命名logType为log_type
            to_key: log_type
    - lowercase_string:
        with_keys:
          - log_type
          - application_name
    - date:
        from_time_received: false
        destination: "@timestamp"
        source: start_time
        date_when: null
  sink:
    - opensearch:
        hosts:
          - "<opensearch-host>" # OpenSearch地址，例如 https://localhost:9200
        index: thinkbridge-trace-${application_name}-${log_type}
        username: "<opensearch-username>" # 用户名
        password: "<opensearch-password>" # 密码

entry-pipeline:
  source:
    rabbitmq:
      host: "<rabbitmq-host>" # Rabbitmq地址
      port: 5672 # Rabbitmq端口
      username: "<rabbitmq-username>" # Rabbitmq用户名
      password: "<rabbitmq-password>" # Rabbitmq密码
      exchange: "<rabbitmq-exchange>" # 交换机名称
      routing_key: "<routing-key>" # 路由键
      queue: "<rabbitmq-queue>" # 队列名称
      durable: true # 队列持久化
  processor:
    - parse_json:
        source: message # 解析rabbitmq原始消息
    - delete_entries:
        with_keys:
          - event
          - "@version"
    - parse_json:
        source: message # 二次解析message字段
    - delete_entries:
        with_keys:
          - message
    - rename_keys:
        entries:
          - from_key: logType # 重命名logType为log_type
            to_key: log_type
    - lowercase_string:
        with_keys:
          - log_type
          - application_name
    - date:
        from_time_received: false
        destination: "@timestamp"
        source: start_time
        date_when: null
  sink:
    - opensearch:
        hosts:
          - "<opensearch-host>" # OpenSearch地址，例如 https://localhost:9200
        index: thinkbridge-trace-${application_name}-${log_type}
        username: "<opensearch-username>" # 用户名
        password: "<opensearch-password>" # 密码

input {
  pulsar {
    service_url => "<pulsar-service-url>" # Pulsar服务地址，例如 pulsar://localhost:6650
    topics => ["<pulsar-topic>"] # 订阅主题
    subscription_name => "<subscription-name>" # 订阅名称
    consumer_name => "logstash-consumer"
    ack_timeout => "30000"
  }
}

filter {
  json {
    source => "message"
    remove_field => ["event","@version"]
  }

  mutate {
      add_field => { "messagetmp" => "%{message}" }
  }
  json {
      source => "messagetmp"
      remove_field => ["messagetmp", "message"]
  }

  mutate {
    rename => { "logType" => "log_type" }
  }

  mutate {
    add_field => { "[@metadata][log_type]" => "%{log_type}" }
    add_field => { "[@metadata][application_name]" => "%{application_name}" }
  }

  mutate {
    lowercase => ["[@metadata][log_type]", "[@metadata][application_name]"]
  }
  mutate {
    remove_field => ["@timestamp"]
    rename => { "start_time" => "@timestamp" }
  }
}

output {
  elasticsearch {
    hosts => ["<elasticsearch-host>"]
    index => "thinkbridge-trace-%{[@metadata][application_name]}-%{[@metadata][log_type]}"
    user => "<elasticsearch-username>"
    password => "<elasticsearch-password>"
  }
}

entry-pipeline:
  source:
    pulsar:
      service_url: "<pulsar-service-url>" # Pulsar服务地址
      topics:
        - name: "<pulsar-topic>" # 订阅主题
      subscription_name: "<subscription-name>" # 订阅名称
  processor:
    - parse_json:
        source: message # 解析pulsar原始消息
    - delete_entries:
        with_keys:
          - event
          - "@version"
    - parse_json:
        source: message # 二次解析message字段
    - delete_entries:
        with_keys:
          - message
    - rename_keys:
        entries:
          - from_key: logType # 重命名logType为log_type
            to_key: log_type
    - lowercase_string:
        with_keys:
          - log_type
          - application_name
    - date:
        from_time_received: false
        destination: "@timestamp"
        source: start_time
        date_when: null
  sink:
    - opensearch:
        hosts:
          - "<opensearch-host>" # OpenSearch地址
        index: thinkbridge-trace-${application_name}-${log_type}
        username: "<opensearch-username>"
        password: "<opensearch-password>"