Bina sistem pengumpulan dan analisis log berdasarkan Spring Boot dan Flume

Memandangkan skala sistem perusahaan terus berkembang, log sistem menjadi lebih besar dan lebih besar Tanpa sistem pengumpulan dan analisis log yang boleh dipercayai, adalah sukar untuk memantau dan menyelenggara sistem dengan berkesan. Artikel ini akan memperkenalkan cara membina sistem pengumpulan dan analisis log yang cekap berdasarkan Spring Boot dan Flume.

1. Prasyarat

Sebelum anda bermula, anda perlu memasang dan menyediakan perisian berikut:

• JDK 8 ke atas
• Maven 3.3 ke atas
• Apache Flume 1.9.0 ke atas
• Elasticsearch 7.6.2 ke atas
• Kibana 7.6.2 ke atas

2. Konfigurasi aplikasi Spring Boot

Pertama, kita perlu mencipta aplikasi Spring Boot dan menambah kebergantungan yang diperlukan:

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-web</artifactId>
</dependency>

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-log4j2</artifactId>
</dependency>

dalam application.properties Dalam fail, tambahkan konfigurasi berikut:

# 应用端口号
server.port=8080

# log4j2配置
logging.config=classpath:log4j2.xml

# flume配置
flume.agentName=myflume
flume.sourceType=avro
flume.clientType=load-balancing
flume.hosts=localhost:41414

# elasticsearch配置
spring.elasticsearch.rest.uris=http://localhost:9200

Dalam konfigurasi di atas, kami menyatakan nombor port aplikasi, fail konfigurasi log4j2, konfigurasi berkaitan Flume dan URI akses Elasticsearch.

3. Pengumpul Log

Untuk menghantar log aplikasi ke Flume, kami perlu mencipta Log4j2 Appender tersuai.

@Plugin(name = "Flume", category = "Core", elementType = "appender", printObject = true)
public class FlumeAppender extends AbstractAppender {

    private static final ObjectMapper MAPPER = new ObjectMapper();
    private final FlumeClient client;
    private final String sourceType;

    protected FlumeAppender(String name, Filter filter, Layout<? extends Serializable> layout,
                            FlumeClient client, String sourceType) {
        super(name, filter, layout, true);
        this.client = client;
        this.sourceType = sourceType;
    }

    @PluginFactory
    public static FlumeAppender createAppender(@PluginAttr("name") String name,
                                               @PluginElement("Filters") Filter filter,
                                               @PluginElement("Layout") Layout<? extends Serializable> layout,
                                               @PluginAttr("sourceType") String sourceType,
                                               @PluginAttr("hosts") String hosts) {
        if (name == null) {
            LOGGER.error("FlumeAppender missing name");
            return null;
        }
        if (client == null) {
            LOGGER.error("FlumeAppender missing client");
            return null;
        }
        return new FlumeAppender(name, filter, layout, createClient(hosts), sourceType);
    }

    private static FlumeClient createClient(String hosts) {
        LoadBalancingRpcClient rpcClient = new LoadBalancingRpcClient();
        String[] hostArray = hosts.split(",");
        for (String host : hostArray) {
            String[] hostParts = host.split(":");
            rpcClient.addHost(new InetSocketAddress(hostParts[0], Integer.parseInt(hostParts[1])));
        }
        Properties props = new Properties();
        props.setProperty(RpcClientConfigurationConstants.CONFIG_CLIENT_TYPE, "default_loadbalance");
        props.setProperty(RpcClientConfigurationConstants.CONFIG_HOSTS, hosts);
        props.setProperty(RpcClientConfigurationConstants.CONFIG_MAX_BACKOFF, "10000");
        AvroEventSerializer serializer = new AvroEventSerializer();
        serializer.configure(props, false);
        return new FlumeClient(rpcClient, serializer);
    }

    @Override
    public void append(LogEvent event) {
        try {
            byte[] body = ((StringLayout) this.getLayout()).toByteArray(event);
            Map<String, String> headers = new HashMap<>();
            headers.put("timestamp", Long.toString(event.getTimeMillis()));
            headers.put("source", "log4j");
            headers.put("sourceType", sourceType);
            Event flumeEvent = EventBuilder.withBody(body, headers);
            client.sendEvent(flumeEvent);
        } catch (Exception e) {
            LOGGER.error("Failed to send event to Flume", e);
        }
    }
}

Dalam kod di atas, kami melaksanakan Log4j2 Appender, yang akan membungkus acara log ke dalam Acara Flume dan menghantarnya ke pelayan Flume.

Buat fail konfigurasi log4j2 dan konfigurasikan FlumeAppender.

<?xml version="1.0" encoding="UTF-8"?>
<Configuration>
    <Appenders>
        <Flume name="flume" sourceType="spring-boot" hosts="${flume.hosts}">
            <PatternLayout pattern="%d{yyyy-MM-dd HH:mm:ss.SSS} [%t] %-5level %logger{36} - %msg%n"/>
        </Flume>
    </Appenders>
    <Loggers>
        <Root level="info">
            <AppenderRef ref="flume"/>
        </Root>
    </Loggers>
</Configuration>

Dalam fail konfigurasi log4j2 ini, kami mentakrifkan FlumeAppender dan merujuknya dalam Root Logger.

4. Konfigurasi Flume

Kami perlu mengkonfigurasi Flume untuk menerima mesej log yang dihantar daripada aplikasi dalam Flume Agent dan menghantarnya ke Elasticsearch.

Buat fail konfigurasi Flume seperti yang ditunjukkan di bawah.

# Define the agent name and the agent sources and sinks
myflume.sources = mysource
myflume.sinks = mysink
myflume.channels = channel1

# Define the source
myflume.sources.mysource.type = avro
myflume.sources.mysource.bind = 0.0.0.0
myflume.sources.mysource.port = 41414

# Define the channel
myflume.channels.channel1.type = memory
myflume.channels.channel1.capacity = 10000
myflume.channels.channel1.transactionCapacity = 1000

# Define the sink
myflume.sinks.mysink.type = org.elasticsearch.hadoop.flume.ElasticsearchSink
myflume.sinks.mysink.hostNames = localhost:9200
myflume.sinks.mysink.indexName = ${type}-%{+YYYY.MM.dd}
myflume.sinks.mysink.batchSize = 1000
myflume.sinks.mysink.typeName = ${type}

# Link the source and sink with the channel
myflume.sources.mysource.channels = channel1
myflume.sinks.mysink.channel = channel1

Dalam fail konfigurasi Flume, kami mentakrifkan ejen, sumber dan sinki. Sumber ialah jenis avro, terikat pada port 41414, saluran1 ialah jenis memori, kapasiti ialah 10000 dan kapasiti transaksi ialah 1000. Sinki ialah jenis ElasticsearchSink yang mencipta jenis indeks bernama pada port 9200 hos tempatan dan menyerahkannya kepada Elasticsearch secara berkelompok apabila 1000 acara dicapai.

5. Konfigurasi Elasticsearch dan Kibana

Akhir sekali, kita perlu mengkonfigurasi Elasticsearch dan Kibana. Dalam Elasticsearch, kita perlu mencipta indeks yang sepadan dengan nama indeks yang ditakrifkan dalam fail konfigurasi Flume.

Di Kibana, kita perlu mencipta skema indeks. Dalam menu utama Kibana, pilih "Pengurusan" dan kemudian "Kibana." Dalam corak indeks Kibana, pilih "Buat Corak Indeks". Masukkan nama indeks yang ditakrifkan dalam fail konfigurasi Flume dan ikut arahan untuk mengkonfigurasinya.

Kami juga perlu mencipta Papan Pemuka untuk Kibana untuk melihat mesej log aplikasi. Dalam menu utama Kibana, pilih "Papan Pemuka" dan kemudian "Buat Papan Pemuka". Dalam tab "Visualisasi", pilih "Tambah visualisasi". Pilih Jadual Data dan konfigurasikan medan dan pilihan visualisasi yang diperlukan.

6. Kesimpulan

Dalam artikel ini, kami memperkenalkan cara menggunakan Spring Boot and Flume untuk membina sistem pengumpulan dan analisis log yang cekap. Kami melaksanakan Appender log4j2 tersuai untuk menghantar peristiwa log aplikasi ke pelayan Flume, dan menggunakan Elasticsearch dan Kibana untuk analisis dan visualisasi log. Saya harap artikel ini dapat membantu anda membina sistem pengumpulan dan analisis log anda sendiri.

Atas ialah kandungan terperinci Bina sistem pengumpulan dan analisis log berdasarkan Spring Boot dan Flume. Untuk maklumat lanjut, sila ikut artikel berkaitan lain di laman web China PHP!