How to Build a Scalable Spring Cloud ELK Logging Pipeline for Microservices

Logging Process Flowchart

This diagram shows the complete log processing flow: each microservice server runs Filebeat, which forwards logs to Logstash; Logstash (without filtering) outputs to Elasticsearch, and logs are visualized via Kibana.

Download Software

Read the following article

Springboot integration with ELK logging detailed steps

This article explains the download and installation of Elasticsearch, Logstash, and Kibana.

Download via the following links

Elasticsearch Configuration

Modify %ES_HOME%\config\elasticsearch.yml to set the cluster name: cluster.name: myes For a single‑node setup no further changes are required; the default port is 9200.

Start Elasticsearch: %ES_HOME%\bin\elasticsearch.bat Verify startup status via the browser.

Kibana Configuration

Edit %KIBANA_HOME%\config\kibana.yml to point to Elasticsearch: elasticsearch.hosts: ["http://localhost:9200"] Start Kibana:

Filebeat Configuration

Edit %FILEBEAT_HOME%\filebeat.yml and define inputs for two services:

filebeat.inputs:
- type: log
  enabled: true
  paths:
    - E:\logs\es-elk\*.log
  tags: ["elk-service1"]
  multiline:
    pattern: '^[0-9]{4}'
    negate: true
    match: after
    timeout: 3s
- type: log
  enabled: true
  paths:
    - E:\logs\ss\*.log
  tags: ["elk-service2"]
  multiline:
    pattern: '^[0-9]{4}'
    negate: true
    match: after
    timeout: 3s

Output to Logstash on port 5044:

output.logstash:
  hosts: ["localhost:5044"]

Start Filebeat:

filebeat.exe -e -c filebeat.yml

Logstash Configuration

Create %LOGSTASH_HOME%\bin\logstash.conf with the following content:

input {
  beats {
    port => 5044
    type => "beats"
  }
}
output {
  if [type] == "beats" and "elk-service1" in [tags] {
    elasticsearch {
      hosts => "localhost:9200"
      index => "elk-service1-%{+YYYY.MM.dd}"
    }
  }
  if [type] == "beats" and "elk-service2" in [tags] {
    elasticsearch {
      hosts => "localhost:9200"
      index => "elk-service2-%{+YYYY.MM.dd}"
    }
  }
}

Start Logstash:

logstash.bat -f logstash.conf

Microservice Logging Configuration

Configure Logback for each service (example for service1):

<configuration scan="true">
    <appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender">
        <encoder>
            <pattern>1-%d{yyyy-MM-dd HH:mm:ss} [%thread] %-5level %logger - %msg%n</pattern>
            <charset>UTF-8</charset>
        </encoder>
    </appender>
    <property name="LOG_PATH" value="../logs"/>
    <property name="SERVICE_NAME" value="es-elk"/>
    <property name="APPDIR" value="es-elk"/>
    <appender name="FILEERROR" class="ch.qos.logback.core.rolling.RollingFileAppender">
        <file>${LOG_PATH}/${APPDIR}/log_error.log</file>
        <rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy">
            <fileNamePattern>${LOG_PATH}/${APPDIR}/log-error-%d{yyyy-MM-dd}.%i.log</fileNamePattern>
            <timeBasedFileNamingAndTriggeringPolicy class="ch.qos.logback.core.rolling.SizeAndTimeBasedFNATP">
                <maxFileSize>20KB</maxFileSize>
            </timeBasedFileNamingAndTriggeringPolicy>
        </rollingPolicy>
        <append>true</append>
        <encoder class="ch.qos.logback.classic.encoder.PatternLayoutEncoder">
            <pattern>%d{yyyy-MM-dd HH:mm:ss.SSS} %-5level %logger Line:%-3L - %msg%n</pattern>
            <charset>utf-8</charset>
        </encoder>
        <filter class="ch.qos.logback.classic.filter.LevelFilter">
            <level>error</level>
            <onMatch>ACCEPT</onMatch>
            <onMismatch>DENY</onMismatch>
        </filter>
    </appender>
    <appender name="FILEINFO" class="ch.qos.logback.core.rolling.RollingFileAppender">
        <file>${LOG_PATH}/${APPDIR}/log_info.log</file>
        <rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy">
            <fileNamePattern>${LOG_PATH}/${APPDIR}/log-info-%d{yyyy-MM-dd}.%i.log</fileNamePattern>
            <timeBasedFileNamingAndTriggeringPolicy class="ch.qos.logback.core.rolling.SizeAndTimeBasedFNATP">
                <maxFileSize>20KB</maxFileSize>
            </timeBasedFileNamingAndTriggeringPolicy>
        </rollingPolicy>
        <append>true</append>
        <encoder class="ch.qos.logback.classic.encoder.PatternLayoutEncoder">
            <pattern>%d{yyyy-MM-dd HH:mm:ss.SSS} %-5level %logger Line:%-3L - %msg%n</pattern>
            <charset>utf-8</charset>
        </encoder>
        <filter class="ch.qos.logback.classic.filter.LevelFilter">
            <level>info</level>
            <onMatch>ACCEPT</onMatch>
            <onMismatch>DENY</onMismatch>
        </filter>
    </appender>
    <appender name="CONSOLE" class="ch.qos.logback.core.ConsoleAppender">
        <encoder>
            <pattern>%d{yyyy-MM-dd HH:mm:ss.SSS} %-5level %logger Line:%-3L - %msg%n</pattern>
            <charset>UTF-8</charset>
        </encoder>
    </appender>
    <logger name="org.springframework.data.mybatis" level="DEBUG"/>
    <logger name="org.springframework.jdbc.core" additivity="false" level="DEBUG">
        <appender-ref ref="CONSOLE"/>
    </logger>
    <springProfile name="dev">
        <root level="INFO">
            <appender-ref ref="FILEINFO"/>
            <appender-ref ref="FILEERROR"/>
        </root>
    </springProfile>
</configuration>

Controller example for service1:

@RestController
@RequestMapping("/elk")
public class ElkController {
    private static Logger logger = LoggerFactory.getLogger(ElkController.class);

    @GetMapping("/index")
    public Object index(String info, Integer a) {
        logger.info("service1, 你输入的是：{}", info);
        if (a == 0) {
            logger.error("service1, 发生错误了: {}", new RuntimeException("数值不正确"));
        }
        return "success";
    }
}

Repeat a similar controller for service2 with tags "elk-service2".

Testing

Start both services, invoke the endpoints, and verify that logs appear in their respective Elasticsearch indices (elk-service1-* and elk-service2-*).

Both indices are created and contain the routed log entries.