ELK Stack（ES+Logstash+Kibana）全链路通关指南

# 创建用户
useradd elk
# 设置密码
passwd elk
# 授权sudo权限
echo "elk ALL=(ALL) NOPASSWD: ALL" >> /etc/sudoers

sudo systemctl stop firewalld
sudo systemctl disable firewalld

sudo sed -i 's/SELINUX=enforcing/SELINUX=disabled/' /etc/selinux/config
source /etc/selinux/config

# 编辑内核参数文件
sudo vim /etc/sysctl.conf
# 添加以下内容
vm.max_map_count=262144  # ES要求的虚拟内存最大值
fs.file-max=65536         # 系统最大文件句柄数
# 生效配置
sudo sysctl -p

# 编辑限制文件
sudo vim /etc/security/limits.conf
# 添加以下内容（针对elk用户）
elk soft nofile 65536
elk hard nofile 65536
elk soft nproc 4096
elk hard nproc 4096

# 切换到elk用户
su - elk
# 下载ES 8.15.0（官网地址：https://www.elastic.co/cn/downloads/elasticsearch）
wget https://artifacts.elastic.co/downloads/elasticsearch/elasticsearch-8.15.0-linux-x86_64.tar.gz
# 解压
tar -zxvf elasticsearch-8.15.0-linux-x86_64.tar.gz
# 重命名（方便操作）
mv elasticsearch-8.15.0 elasticsearch

# 集群名称（默认elasticsearch，集群部署时所有节点需一致）
cluster.name:elk-cluster
# 节点名称（单节点随意，集群部署时需唯一）
node.name:node-1
# 数据存储目录（建议单独挂载磁盘，避免占用系统盘）
path.data:/home/elk/elasticsearch/data
# 日志存储目录
path.logs:/home/elk/elasticsearch/logs
# 绑定地址（0.0.0.0允许外部访问，开发环境使用，生产环境建议指定具体IP）
network.host:0.0.0.0
# HTTP端口（默认9200）
http.port:9200
# 集群初始化节点（单节点为当前节点）
cluster.initial_master_nodes:["node-1"]
# 关闭HTTPS（开发环境，生产环境建议开启）
xpack.security.enabled:false
xpack.security.http.ssl.enabled:false
# 允许跨域（方便前端/Kibana访问）
http.cors.enabled:true
http.cors.allow-origin:"*"

# 进入ES目录
cd /home/elk/elasticsearch
# 后台启动
./bin/elasticsearch -d

# 访问ES健康检查接口
curl http://localhost:9200/_cat/health?v

epoch      timestamp cluster       status node.total node.data shards pri relo init unassign pending_tasks max_task_wait_time active_shards_percent
1699876532 10:35:32  elk-cluster   green           1         1      0   0    0    0        0             0                  -                100.0%

# 下载Logstash 8.15.0
wget https://artifacts.elastic.co/downloads/logstash/logstash-8.15.0-linux-x86_64.tar.gz
# 解压
tar -zxvf logstash-8.15.0-linux-x86_64.tar.gz
# 重命名
mv logstash-8.15.0 logstash

# 输入插件：从MySQL采集数据
input {
  jdbc {
    # MySQL驱动路径（需提前下载mysql-connector-java-8.0.33.jar，放入logstash/lib目录）
    jdbc_driver_library => "/home/elk/logstash/lib/mysql-connector-java-8.0.33.jar"
    # MySQL驱动类名
    jdbc_driver_class => "com.mysql.cj.jdbc.Driver"
    # MySQL连接地址（数据库名：demo，字符集：utf8mb4）
    jdbc_connection_string => "jdbc:mysql://localhost:3306/demo?useSSL=false&serverTimezone=Asia/Shanghai&characterEncoding=utf8mb4"
    # MySQL用户名
    jdbc_user => "root"
    # MySQL密码
    jdbc_password => "root123456"
    # 开启分页查询
    jdbc_paging_enabled => true
    # 每页查询条数
    jdbc_page_size => 1000
    # 定时任务：每10秒执行一次（语法：分 时 日 月 周）
    schedule => "*/10 * * * *"
    # 查询SQL（查询demo表中id大于上次记录的新数据，实现增量同步）
    statement => "SELECT id, username, age, email, create_time FROM user WHERE id > :sql_last_value"
    # 记录上次查询的id值（存储在logstash/data目录下）
    use_column_value => true
    # 作为增量标识的字段（id为自增主键）
    tracking_column => "id"
    # 增量标识字段类型（numeric：数字，timestamp：时间戳）
    tracking_column_type => "numeric"
    # 初始化时的默认值（首次执行时查询id>0的所有数据）
    last_run_metadata_path => "/home/elk/logstash/data/sql_last_value.txt"
  }
}

# 过滤插件：数据清洗与转换
filter {
  # 字段类型转换（将age从字符串转为整数，create_time转为时间类型）
  mutate {
    convert => { "age" => "integer" }
  }
  # 时间格式转换（将MySQL的datetime格式转为ES支持的ISO8601格式）
  date {
    match => { "create_time" => "yyyy-MM-dd HH:mm:ss" }
    target => "@timestamp"  # 覆盖ES默认的@timestamp字段
  }
  # 移除无用字段（如logstash自动添加的@version）
  mutate {
    remove_field => ["@version"]
  }
}

# 输出插件：将处理后的数据写入ES
output {
  elasticsearch {
    # ES连接地址
    hosts => ["http://localhost:9200"]
    # 写入ES的索引名（用户数据：user-index）
    index => "user-index"
    # 用MySQL的id作为ES文档的_id（避免重复数据）
    document_id => "%{id}"
  }
  # 调试输出：将处理后的数据打印到控制台
  stdout {
    codec => rubydebug
  }
}

cd /home/elk/logstash
# 指定配置文件启动
./bin/logstash -f config/mysql-to-es.conf

CREATE DATABASEIFNOTEXISTS demo DEFAULTCHARACTERSET utf8mb4 COLLATE utf8mb4_unicode_ci;
USE demo;
CREATETABLEIFNOTEXISTSuser (
idBIGINT AUTO_INCREMENT PRIMARY KEYCOMMENT'主键ID',
  username VARCHAR(50) NOTNULLCOMMENT'用户名',
  age INTCOMMENT'年龄',
  email VARCHAR(100) COMMENT'邮箱',
  create_time DATETIME NOTNULLDEFAULTCURRENT_TIMESTAMPCOMMENT'创建时间'
) COMMENT'用户表';

-- 插入测试数据
INSERTINTOuser (username, age, email) VALUES ('zhangsan', 25, 'zhangsan@demo.com'), ('lisi', 30, 'lisi@demo.com');

curl http://localhost:9200/user-index/_search?q=*

# 下载Kibana 8.15.0
wget https://artifacts.elastic.co/downloads/kibana/kibana-8.15.0-linux-x86_64.tar.gz
# 解压
tar -zxvf kibana-8.15.0-linux-x86_64.tar.gz
# 重命名
mv kibana-8.15.0 kibana

# 服务器端口（默认5601）
server.port: 5601
# 绑定地址（0.0.0.0允许外部访问）
server.host: "0.0.0.0"
# ES连接地址（需与ES配置一致）
elasticsearch.hosts: ["http://localhost:9200"]
# 界面语言（中文）
i18n.locale: "zh-CN"

cd /home/elk/kibana
# 后台启动
nohup ./bin/kibana &

curl -X PUT "http://localhost:9200/product-index" -H "Content-Type: application/json" -d '
{
  "settings": {
    "number_of_shards": 3,    # 主分片数量3
    "number_of_replicas": 1   # 副本分片数量1
  },
  "mappings": {
    "properties": {
      "product_id": { "type": "long", "index": true },  # 商品ID，长整型，可索引
      "product_name": { "type": "text", "analyzer": "ik_max_word" },  # 商品名称，文本类型，使用IK分词器
      "category": { "type": "keyword" },  # 商品分类，关键字类型，精确匹配
      "price": { "type": "double" },  # 商品价格，双精度浮点型
      "create_time": { "type": "date", "format": "yyyy-MM-dd HH:mm:ss" }  # 创建时间，指定格式
    }
  }
}'

curl -X PUT "http://localhost:9200/product-index/_doc/1001" -H "Content-Type: application/json" -d '
{
  "product_id": 1001,
  "product_name": "华为Mate 60 Pro",
  "category": "手机",
  "price": 6999.00,
  "create_time": "2024-01-15 10:30:00"
}'

curl -X GET "http://localhost:9200/product-index/_doc/1001"

curl -X GET "http://localhost:9200/product-index/_search" -H "Content-Type: application/json" -d '
{
  "query": {
    "match": {
      "product_name": "华为"
    }
  }
}'

curl -X GET "http://localhost:9200/product-index/_search" -H "Content-Type: application/json" -d '
{
  "size": 0,  # 不返回具体文档，只返回聚合结果
  "aggs": {
    "category_count": {
      "terms": {
        "field": "category"  # 按category字段分组
      }
    }
  }
}'

curl -X DELETE "http://localhost:9200/product-index"

<dependencies>
    <!-- Elasticsearch Java客户端 -->
    <dependency>
        <groupId>co.elastic.clients</groupId>
        <artifactId>elasticsearch-java</artifactId>
        <version>8.15.0</version>
    </dependency>
    <!--  Jackson依赖（客户端依赖Jackson处理JSON） -->
    <dependency>
        <groupId>com.fasterxml.jackson.core</groupId>
        <artifactId>jackson-databind</artifactId>
        <version>2.15.2</version>
    </dependency>
    <!-- Lombok（日志、Getter/Setter） -->
    <dependency>
        <groupId>org.projectlombok</groupId>
        <artifactId>lombok</artifactId>
        <version>1.18.30</version>
        <scope>provided</scope>
    </dependency>
    <!-- Spring Boot Test（测试用） -->
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-test</artifactId>
        <version>3.1.5</version>
        <scope>test</scope>
    </dependency>
</dependencies>

package com.jam.demo.es;

import co.elastic.clients.elasticsearch.ElasticsearchClient;
import co.elastic.clients.json.jackson.JacksonJsonpMapper;
import co.elastic.clients.transport.ElasticsearchTransport;
import co.elastic.clients.transport.rest_client.RestClientTransport;
import lombok.extern.slf4j.Slf4j;
import org.apache.http.HttpHost;
import org.elasticsearch.client.RestClient;
import org.springframework.util.ObjectUtils;

import java.io.Closeable;
import java.io.IOException;

/**
 * Elasticsearch客户端工具类
 * 功能：创建并管理ElasticsearchClient实例
 * @author ken
 */
@Slf4j
publicclass EsClientUtil implements Closeable {

    privatestatic ElasticsearchClient esClient;
    privatestatic RestClient restClient;
    privatestatic ElasticsearchTransport transport;

    /**
     * 初始化Elasticsearch客户端
     * @param host ES主机地址
     * @param port ES端口
     * @return ElasticsearchClient实例
     */
    public static ElasticsearchClient initClient(String host, int port) {
        if (ObjectUtils.isEmpty(esClient)) {
            // 1. 创建RestClient
            restClient = RestClient.builder(new HttpHost(host, port)).build();
            // 2. 创建传输层（使用Jackson处理JSON）
            transport = new RestClientTransport(restClient, new JacksonJsonpMapper());
            // 3. 创建ElasticsearchClient
            esClient = new ElasticsearchClient(transport);
            log.info("Elasticsearch客户端初始化成功，host：{}，port：{}", host, port);
        }
        return esClient;
    }

    /**
     * 关闭客户端资源
     */
    @Override
    public void close() throws IOException {
        if (!ObjectUtils.isEmpty(transport)) {
            transport.close();
        }
        if (!ObjectUtils.isEmpty(restClient)) {
            restClient.close();
        }
        log.info("Elasticsearch客户端资源已关闭");
    }
}

package com.jam.demo.es.entity;

import lombok.Data;
import java.time.LocalDateTime;

/**
 * 商品实体类（对应ES中的product-index索引）
 * @author ken
 */
@Data
publicclass Product {
    /**
     * 商品ID（对应ES文档的product_id字段）
     */
    private Long productId;
    /**
     * 商品名称（对应ES文档的product_name字段）
     */
    private String productName;
    /**
     * 商品分类（对应ES文档的category字段）
     */
    private String category;
    /**
     * 商品价格（对应ES文档的price字段）
     */
    private Double price;
    /**
     * 创建时间（对应ES文档的create_time字段）
     */
    private LocalDateTime createTime;
}

package com.jam.demo.es;

import co.elastic.clients.elasticsearch.ElasticsearchClient;
import co.elastic.clients.elasticsearch._types.query_dsl.MatchQuery;
import co.elastic.clients.elasticsearch._types.query_dsl.Query;
import co.elastic.clients.elasticsearch.core.*;
import co.elastic.clients.elasticsearch.core.bulk.BulkOperation;
import co.elastic.clients.elasticsearch.core.bulk.CreateOperation;
import co.elastic.clients.elasticsearch.indices.CreateIndexRequest;
import co.elastic.clients.elasticsearch.indices.CreateIndexResponse;
import co.elastic.clients.elasticsearch.indices.DeleteIndexRequest;
import com.jam.demo.es.entity.Product;
import lombok.extern.slf4j.Slf4j;
import org.springframework.util.CollectionUtils;
import org.springframework.util.ObjectUtils;
import org.springframework.util.StringUtils;

import java.io.IOException;
import java.util.List;
import java.util.stream.Collectors;

/**
 * 商品ES操作服务类
 * 功能：实现商品索引的创建、文档的CRUD、检索、聚合等操作
 * @author ken
 */
@Slf4j
publicclass EsProductService {

    privatefinal ElasticsearchClient esClient;
    privatestaticfinal String PRODUCT_INDEX = "product-index";

    /**
     * 构造方法（初始化ES客户端）
     * @param host ES主机地址
     * @param port ES端口
     */
    public EsProductService(String host, int port) {
        this.esClient = EsClientUtil.initClient(host, port);
    }

    /**
     * 创建商品索引（指定映射）
     * @return 是否创建成功
     * @throws IOException ES操作异常
     */
    public boolean createProductIndex() throws IOException {
        // 构建索引映射配置
        String mapping = "{\n" +
                "  \"settings\": {\n" +
                "    \"number_of_shards\": 3,\n" +
                "    \"number_of_replicas\": 1\n" +
                "  },\n" +
                "  \"mappings\": {\n" +
                "    \"properties\": {\n" +
                "      \"productId\": { \"type\": \"long\", \"index\": true },\n" +
                "      \"productName\": { \"type\": \"text\", \"analyzer\": \"ik_max_word\" },\n" +
                "      \"category\": { \"type\": \"keyword\" },\n" +
                "      \"price\": { \"type\": \"double\" },\n" +
                "      \"createTime\": { \"type\": \"date\", \"format\": \"yyyy-MM-dd HH:mm:ss\" }\n" +
                "    }\n" +
                "  }\n}";

        CreateIndexRequest request = CreateIndexRequest.of(builder -> builder
                .index(PRODUCT_INDEX)
                .withJson(mapping.getBytes())
        );

        CreateIndexResponse response = esClient.indices().create(request);
        log.info("创建商品索引结果：{}，索引名：{}", response.acknowledged(), PRODUCT_INDEX);
        return response.acknowledged();
    }

    /**
     * 插入单个商品文档
     * @param product 商品实体
     * @param docId ES文档ID
     * @return 是否插入成功
     * @throws IOException ES操作异常
     */
    public boolean insertProductDoc(Product product, String docId) throws IOException {
        if (ObjectUtils.isEmpty(product) || StringUtils.isEmpty(docId)) {
            log.error("插入商品文档失败：商品信息或文档ID为空");
            returnfalse;
        }

        IndexRequest<Product> request = IndexRequest.of(builder -> builder
                .index(PRODUCT_INDEX)
                .id(docId)
                .document(product)
        );

        IndexResponse response = esClient.index(request);
        log.info("插入商品文档结果：{}，文档ID：{}", response.result(), docId);
        return"created".equals(response.result().jsonValue()) || "updated".equals(response.result().jsonValue());
    }

    /**
     * 批量插入商品文档
     * @param productList 商品列表
     * @return 批量操作结果（成功数量）
     * @throws IOException ES操作异常
     */
    public int bulkInsertProductDocs(List<Product> productList) throws IOException {
        if (CollectionUtils.isEmpty(productList)) {
            log.error("批量插入商品文档失败：商品列表为空");
            return0;
        }

        // 构建批量操作列表
        List<BulkOperation> operations = productList.stream()
                .map(product -> {
                    // 用商品ID作为文档ID
                    String docId = product.getProductId().toString();
                    return BulkOperation.of(builder -> builder
                            .create(CreateOperation.of(cBuilder -> cBuilder
                                    .index(PRODUCT_INDEX)
                                    .id(docId)
                                    .document(product)
                            ))
                    );
                })
                .collect(Collectors.toList());

        BulkRequest request = BulkRequest.of(builder -> builder.operations(operations));
        BulkResponse response = esClient.bulk(request);

        if (response.errors()) {
            log.error("批量插入商品文档存在失败记录");
            response.items().forEach(item -> {
                if (item.error() != null) {
                    log.error("文档ID：{}，插入失败：{}", item.id(), item.error().reason());
                }
            });
        }

        int successCount = (int) response.items().stream()
                .filter(item -> item.error() == null)
                .count();
        log.info("批量插入商品文档完成，总数量：{}，成功数量：{}", productList.size(), successCount);
        return successCount;
    }

    /**
     * 根据文档ID查询商品
     * @param docId 文档ID
     * @return 商品实体
     * @throws IOException ES操作异常
     */
    public Product getProductDocById(String docId) throws IOException {
        if (StringUtils.isEmpty(docId)) {
            log.error("查询商品文档失败：文档ID为空");
            returnnull;
        }

        GetRequest request = GetRequest.of(builder -> builder
                .index(PRODUCT_INDEX)
                .id(docId)
        );

        GetResponse<Product> response = esClient.get(request, Product.class);
        if (response.found()) {
            log.info("查询商品文档成功，文档ID：{}", docId);
            return response.source();
        } else {
            log.info("未查询到商品文档，文档ID：{}", docId);
            returnnull;
        }
    }

    /**
     * 全文检索商品（根据商品名称）
     * @param productName 商品名称关键词
     * @return 商品列表
     * @throws IOException ES操作异常
     */
    public List<Product> searchProductByTitle(String productName) throws IOException {
        if (StringUtils.isEmpty(productName)) {
            log.error("检索商品失败：商品名称关键词为空");
            returnnull;
        }

        // 构建match查询（全文检索）
        Query query = MatchQuery.of(mBuilder -> mBuilder
                .field("productName")
                .query(productName)
        )._toQuery();

        SearchRequest request = SearchRequest.of(sBuilder -> sBuilder
                .index(PRODUCT_INDEX)
                .query(query)
        );

        SearchResponse<Product> response = esClient.search(request, Product.class);
        List<Product> productList = response.hits().hits().stream()
                .map(hit -> hit.source())
                .collect(Collectors.toList());

        log.info("检索商品完成，关键词：{}，匹配数量：{}", productName, productList.size());
        return productList;
    }

    /**
     * 删除商品索引
     * @return 是否删除成功
     * @throws IOException ES操作异常
     */
    public boolean deleteProductIndex() throws IOException {
        DeleteIndexRequest request = DeleteIndexRequest.of(builder -> builder.index(PRODUCT_INDEX));
        esClient.indices().delete(request);
        log.info("删除商品索引成功，索引名：{}", PRODUCT_INDEX);
        returntrue;
    }
}

package com.jam.demo.es;

import com.jam.demo.es.entity.Product;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import java.io.IOException;
import java.time.LocalDateTime;
import java.util.Arrays;
import java.util.List;
importstatic org.junit.Assert.*;

/**
 * 商品ES操作服务测试类
 * @author ken
 */
publicclass EsProductServiceTest {

    private EsProductService esProductService;
    private EsClientUtil esClientUtil;

    @Before
    public void init() {
        // 初始化ES客户端和服务类（本地ES，端口9200）
        esClientUtil = new EsClientUtil();
        esProductService = new EsProductService("localhost", 9200);
    }

    @Test
    public void testProductIndexOperation() throws IOException {
        // 1. 创建商品索引
        boolean createResult = esProductService.createProductIndex();
        assertTrue(createResult);

        // 2. 插入单个商品文档
        Product product1 = new Product();
        product1.setProductId(1001L);
        product1.setProductName("华为Mate 60 Pro");
        product1.setCategory("手机");
        product1.setPrice(6999.00);
        product1.setCreateTime(LocalDateTime.of(2024, 1, 15, 10, 30, 0));
        boolean insertSingleResult = esProductService.insertProductDoc(product1, "1001");
        assertTrue(insertSingleResult);

        // 3. 批量插入商品文档
        Product product2 = new Product();
        product2.setProductId(1002L);
        product2.setProductName("苹果iPhone 15");
        product2.setCategory("手机");
        product2.setPrice(7999.00);
        product2.setCreateTime(LocalDateTime.of(2024, 2, 20, 14, 15, 0));

        Product product3 = new Product();
        product3.setProductId(1003L);
        product3.setProductName("小米笔记本Pro");
        product3.setCategory("笔记本电脑");
        product3.setPrice(5999.00);
        product3.setCreateTime(LocalDateTime.of(2024, 3, 10, 9, 45, 0));

        List<Product> productList = Arrays.asList(product2, product3);
        int bulkInsertCount = esProductService.bulkInsertProductDocs(productList);
        assertEquals(2, bulkInsertCount);

        // 4. 根据ID查询商品
        Product queryProduct = esProductService.getProductDocById("1001");
        assertNotNull(queryProduct);
        assertEquals("华为Mate 60 Pro", queryProduct.getProductName());

        // 5. 全文检索商品
        List<Product> searchProducts = esProductService.searchProductByTitle("手机");
        assertNotNull(searchProducts);
        assertEquals(2, searchProducts.size());

        // 6. 删除商品索引（测试完成后清理）
        boolean deleteResult = esProductService.deleteProductIndex();
        assertTrue(deleteResult);
    }

    @After
    public void destroy() throws IOException {
        // 关闭客户端资源
        esClientUtil.close();
    }
}

# 进入ES插件目录
cd /home/elk/elasticsearch/plugins
# 创建ik目录
mkdir ik
# 下载IK分词器（官网地址：https://github.com/medcl/elasticsearch-analysis-ik/releases）
wget https://github.com/medcl/elasticsearch-analysis-ik/releases/download/v8.15.0/elasticsearch-analysis-ik-8.15.0.zip
# 解压到ik目录
unzip elasticsearch-analysis-ik-8.15.0.zip -d ik
# 删除压缩包
rm -rf elasticsearch-analysis-ik-8.15.0.zip

# 查看ES插件列表
/home/elk/elasticsearch/bin/elasticsearch-plugin list
# 输出“analysis-ik”表示安装成功

# ik_max_word模式
curl -X POST "http://localhost:9200/_analyze" -H "Content-Type: application/json" -d '
{
  "analyzer": "ik_max_word",
  "text": "华为Mate 60 Pro手机"
}'

# ik_smart模式
curl -X POST "http://localhost:9200/_analyze" -H "Content-Type: application/json" -d '
{
  "analyzer": "ik_smart",
  "text": "华为Mate 60 Pro手机"
}'

华为Mate 60 Pro
小米笔记本Pro

<properties>
    <comment>IK Analyzer 扩展配置</comment>
    <!-- 自定义扩展词典 -->
    <entry key="ext_dict">custom.dic</entry>
    <!-- 自定义停用词词典 -->
    <entry key="ext_stopwords">stopword.dic</entry>
</properties>

curl -X POST "http://localhost:9200/_analyze" -H "Content-Type: application/json" -d '
{
  "analyzer": "ik_max_word",
  "text": "华为Mate 60 Pro手机"
}'

input {
  file {
    path => ["/var/log/application/*.log"]  # 日志文件路径（支持通配符）
    start_position => "beginning"  # 从文件开头读取（默认end：从文件末尾）
    sincedb_path => "/home/elk/logstash/data/sincedb"  # 记录文件读取位置（避免重复读取）
    stat_interval => 1  # 检查文件变化的时间间隔（秒）
    codec => "json"  # 日志格式（JSON格式，若为普通文本可省略）
  }
}

input {
  kafka {
    bootstrap_servers => "kafka-node1:9092,kafka-node2:9092"  # Kafka集群地址
    topics => ["app-log-topic"]  # 要消费的主题
    group_id => "logstash-consumer-group"  # 消费者组ID
    auto_offset_reset => "latest"  # 无偏移量时从最新消息开始消费
    consumer_threads => 5  # 消费线程数
  }
}

filter {
  grok {
    # 匹配Nginx日志格式（日志格式：$remote_addr - $remote_user [$time_local] "$request" $status $body_bytes_sent "$http_referer" "$http_user_agent" "$http_x_forwarded_for" $request_time）
    match => { "message" => "%{IP:client_ip} - %{DATA:remote_user} \[%{HTTPDATE:request_time}\] \"%{HTTP_METHOD:request_method} %{URIPATH:request_path}(%{URIPARAM:request_params})? HTTP/%{NUMBER:http_version}\" %{NUMBER:status:int} %{NUMBER:body_bytes_sent:int} \"%{DATA:http_referer}\" \"%{DATA:user_agent}\" \"%{DATA:http_x_forwarded_for}\" %{NUMBER:request_duration:float}" }
    # 匹配失败时添加标签
    tag_on_failure => ["grok_parse_failure"]
    # 移除原始message字段（解析后无需保留）
    remove_field => ["message"]
  }
}

filter {
  mutate {
    # 1. 类型转换（将status从字符串转为整数，request_duration转为浮点型）
    convert => { "status" => "integer" "request_duration" => "float" }
    # 2. 重命名字段（将http_x_forwarded_for改为forwarded_ip）
    rename => { "http_x_forwarded_for" => "forwarded_ip" }
    # 3. 添加新字段（固定值字段）
    add_field => { "service_name" => "nginx" "env" => "prod" }
    # 4. 移除无用字段
    remove_field => ["remote_user", "http_referer"]
    # 5. 字符串替换（将user_agent中的空字符替换为-）
    gsub => [ "user_agent", " ", "-" ]
  }
}

filter {
  date {
    # 匹配日志中的时间字段（request_time为Nginx日志中的时间，格式：dd/MMM/yyyy:HH:mm:ss Z）
    match => { "request_time" => "dd/MMM/yyyy:HH:mm:ss Z" }
    # 目标字段（覆盖ES默认的@timestamp字段）
    target => "@timestamp"
    # 时间 zone（避免时区偏移）
    timezone => "Asia/Shanghai"
    # 匹配失败时的默认时间（当前时间）
    fallback_to_current_time => true
  }
}

filter {
  # 丢弃状态码为404的日志
  if [status] == 404 {
    drop { }
  }
  # 丢弃测试环境的日志
  if [env] == "test" {
    drop { }
  }
}

filter {
  geoip {
    # 要解析的IP字段（client_ip为Grok提取的客户端IP）
    source => "client_ip"
    # 存储地理位置信息的字段名（默认geoip）
    target => "geoip"
    # 数据库路径（默认使用插件内置的GeoLite2数据库，需定期更新）
    database => "/home/elk/logstash/config/GeoLite2-City.mmdb"
    # 要获取的地理位置字段（默认全量，可指定减少数据量）
    fields => ["country_name", "region_name", "city_name", "location"]
  }
}

output {
  elasticsearch {
    hosts => ["http://localhost:9200"]  # ES集群地址（多个用逗号分隔）
    index => "nginx-access-log-%{+YYYY.MM.dd}"  # 按天生成索引（日志按天分割，便于管理）
    document_id => "%{[@timestamp]}-%{client_ip}-%{request_path}"  # 自定义文档ID，避免重复
    # 批量写入优化（提升性能）
    flush_size => 5000  # 每5000条数据批量提交一次
    idle_flush_time => 10  # 若10秒内未达到flush_size，也批量提交
    # 认证配置（生产环境ES开启安全验证时启用）
    user => "elastic"  # ES默认管理员用户
    password => "elastic123"  # ES用户密码
    # 索引模板自动加载（提前创建索引模板，定义映射）
    template => "/home/elk/logstash/config/nginx-template.json"
    template_name => "nginx-access-template"
    template_overwrite => true  # 覆盖已存在的模板
  }
}

{
  "index_patterns": ["nginx-access-log-*"],  # 匹配所有nginx访问日志索引
"settings": {
    "number_of_shards": 3,
    "number_of_replicas": 1
  },
"mappings": {
    "properties": {
      "client_ip": { "type": "ip" },
      "request_time": { "type": "date", "format": "dd/MMM/yyyy:HH:mm:ss Z" },
      "request_method": { "type": "keyword" },
      "request_path": { "type": "keyword" },
      "status": { "type": "integer" },
      "request_duration": { "type": "float" },
      "user_agent": { "type": "text", "analyzer": "ik_smart" },
      "geoip": {
        "properties": {
          "country_name": { "type": "keyword" },
          "region_name": { "type": "keyword" },
          "city_name": { "type": "keyword" },
          "location": { "type": "geo_point" }  # 地理位置类型，支持地图可视化
        }
      }
    }
  }
}

output {
  kafka {
    bootstrap_servers => "kafka-node1:9092,kafka-node2:9092"  # Kafka集群地址
    topic_id => "processed-nginx-log"  # 目标主题
    # 分区策略（按client_ip哈希分区，保证同一IP的日志进入同一分区）
    partitioner => "org.apache.kafka.clients.producer.internals.DefaultPartitioner"
    partition_key => "%{client_ip}"
    # 批量发送优化
    batch_size => 16384  # 批量发送大小（字节）
    linger_ms => 5  # 等待5毫秒，凑够批量大小再发送
    # 序列化方式（JSON格式）
    codec => "json"
  }
}

output {
  file {
    path => "/home/elk/logstash/output/nginx-log-%{+YYYY.MM.dd}.log"  # 按天生成备份文件
    codec => "json_lines"  # 每行一个JSON对象
    flush_interval => 5  # 每5秒刷新一次缓冲区
    gzip => true  # 启用Gzip压缩，减少磁盘占用
  }
}

output {
  # 当请求响应时间超过5秒时，发送邮件告警
  if [request_duration] > 5 {
    email {
      to => "admin@demo.com"  # 接收告警邮件地址
      from => "logstash-alert@demo.com"  # 发送邮件地址
      subject => "【告警】Nginx请求响应时间过长"  # 邮件主题
      body => "客户端IP：%{client_ip}\n请求路径：%{request_path}\n响应时间：%{request_duration}秒\n发生时间：%{[@timestamp]}"  # 邮件内容
      address => "smtp.demo.com"  # SMTP服务器地址
      port => 465  # SMTP端口（SSL）
      username => "logstash-alert@demo.com"  # SMTP用户名
      password => "alert123"  # SMTP密码
      use_tls => true  # 启用TLS加密
    }
  }
}

# 输入：读取Nginx访问日志
input {
  file {
    path => ["/var/log/nginx/access.log"]
    start_position => "beginning"
    sincedb_path => "/home/elk/logstash/data/sincedb_nginx"
    stat_interval => 1
  }
}

# 过滤：解析、清洗、转换数据
filter {
  # 1. Grok解析Nginx日志字段
  grok {
    match => { "message" => "%{IP:client_ip} - %{DATA:remote_user} \[%{HTTPDATE:request_time}\] \"%{HTTP_METHOD:request_method} %{URIPATH:request_path}(%{URIPARAM:request_params})? HTTP/%{NUMBER:http_version}\" %{NUMBER:status:int} %{NUMBER:body_bytes_sent:int} \"%{DATA:http_referer}\" \"%{DATA:user_agent}\" \"%{DATA:forwarded_ip}\" %{NUMBER:request_duration:float}" }
    tag_on_failure => ["grok_parse_failure"]
    remove_field => ["message", "remote_user", "http_referer"]
  }

  # 2. 时间格式标准化
  date {
    match => { "request_time" => "dd/MMM/yyyy:HH:mm:ss Z" }
    target => "@timestamp"
    timezone => "Asia/Shanghai"
    fallback_to_current_time => true
  }

  # 3. 字段加工
  mutate {
    rename => { "forwarded_ip" => "x_forwarded_ip" }
    add_field => { "service" => "nginx" "env" => "prod" }
  }

  # 4. 地理位置解析
  geoip {
    source => "client_ip"
    target => "geoip"
    fields => ["country_name", "region_name", "city_name", "location"]
  }

  # 5. 丢弃404日志
  if [status] == 404 {
    drop { }
  }
}

# 输出：写入ES（按天分区索引）
output {
  elasticsearch {
    hosts => ["http://localhost:9200"]
    index => "nginx-access-log-%{+YYYY.MM.dd}"
    document_id => "%{[@timestamp]}-%{client_ip}-%{request_path}"
    flush_size => 5000
    idle_flush_time => 10
    template => "/home/elk/logstash/config/nginx-template.json"
    template_name => "nginx-access-template"
    template_overwrite => true
  }

  # 调试输出：控制台打印（生产环境可注释）
  stdout {
    codec => rubydebug
  }
}

sudo chmod 644 /var/log/nginx/access.log
sudo chown elk:elk /var/log/nginx/access.log

cd /home/elk/logstash
./bin/logstash -f config/nginx-log-to-es.conf

# 查询当天的Nginx日志索引
curl http://localhost:9200/nginx-access-log-$(date +%Y.%m.%d)/_search?q=*

- pipeline.id:nginx-log-pipeline
path.config:"/home/elk/logstash/config/nginx-log-to-es.conf"
pipeline.workers:4# 工作线程数（建议等于CPU核心数）
pipeline.batch.size:1000# 每个线程批量处理条数

-pipeline.id:app-log-pipeline
path.config:"/home/elk/logstash/config/app-log-to-es.conf"
pipeline.workers:4
pipeline.batch.size:1000

-Xms4g
-Xmx4g

{
  "query": {
    "bool": {
      "must": [
        { "match": { "status": 200 } },
        { "range": { "request_duration": { "gt": 2 } } }
      ]
    }
  }
}

server.maxPayloadBytes: 10485760  # 最大请求 payload 大小（10MB）
elasticsearch.requestTimeout: 30000  # ES 请求超时时间（30秒）

<dependencies>
    <!-- Spring Boot Web -->
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-web</artifactId>
        <version>3.1.5</version>
    </dependency>

    <!-- MyBatis-Plus -->
    <dependency>
        <groupId>com.baomidou</groupId>
        <artifactId>mybatis-plus-boot-starter</artifactId>
        <version>3.5.4</version>
    </dependency>

    <!-- MySQL驱动 -->
    <dependency>
        <groupId>com.mysql</groupId>
        <artifactId>mysql-connector-j</artifactId>
        <version>8.0.33</version>
        <scope>runtime</scope>
    </dependency>

    <!-- Lombok -->
    <dependency>
        <groupId>org.projectlombok</groupId>
        <artifactId>lombok</artifactId>
        <version>1.18.30</version>
        <scope>provided</scope>
    </dependency>

    <!-- Fastjson2 -->
    <dependency>
        <groupId>com.alibaba.fastjson2</groupId>
        <artifactId>fastjson2</artifactId>
        <version>2.0.32</version>
    </dependency>

    <!-- Swagger3 -->
    <dependency>
        <groupId>org.springdoc</groupId>
        <artifactId>springdoc-openapi-starter-webmvc-ui</artifactId>
        <version>2.1.0</version>
    </dependency>

    <!-- Spring Boot Test -->
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-test</artifactId>
        <version>3.1.5</version>
        <scope>test</scope>
    </dependency>
</dependencies>

spring:
  application:
    name:elk-demo-app
datasource:
    url:jdbc:mysql://localhost:3306/demo?useSSL=false&serverTimezone=Asia/Shanghai&characterEncoding=utf8mb4
    username:root
    password:root123456
    driver-class-name:com.mysql.cj.jdbc.Driver

mybatis-plus:
mapper-locations:classpath:mapper/*.xml
type-aliases-package:com.jam.demo.entity
configuration:
    map-underscore-to-camel-case:true# 下划线转驼峰
    log-impl:org.apache.ibatis.logging.stdout.StdOutImpl# 打印MyBatis日志

# 日志配置（输出JSON格式日志，便于Logstash解析）
logging:
file:
    name:/var/log/elk-demo-app/app.log# 日志文件路径
pattern:
    file:'{"timestamp":"%d{yyyy-MM-dd HH:mm:ss.SSS}","level":"%p","thread":"%t","class":"%c{1}","message":"%m","exception":"%ex{full}"}'# JSON格式
level:
    root:info
    com.jam.demo:debug

# Swagger3配置
springdoc:
api-docs:
    path:/api-docs
swagger-ui:
    path:/swagger-ui.html
    operationsSorter:method
packages-to-scan:com.jam.demo.controller

package com.jam.demo.entity;

import com.baomidou.mybatisplus.annotation.IdType;
import com.baomidou.mybatisplus.annotation.TableId;
import com.baomidou.mybatisplus.annotation.TableName;
import lombok.Data;
import java.time.LocalDateTime;

/**
 * 用户实体类
 * @author ken
 */
@Data
@TableName("user")
publicclass User {
    /**
     * 主键ID
     */
    @TableId(type = IdType.AUTO)
    private Long id;

    /**
     * 用户名
     */
    private String username;

    /**
     * 年龄
     */
    private Integer age;

    /**
     * 邮箱
     */
    private String email;

    /**
     * 创建时间
     */
    private LocalDateTime createTime;
}

package com.jam.demo.mapper;

import com.baomidou.mybatisplus.core.mapper.BaseMapper;
import com.jam.demo.entity.User;
import org.springframework.stereotype.Repository;

/**
 * 用户Mapper接口
 * @author ken
 */
@Repository
public interface UserMapper extends BaseMapper<User> {
}

package com.jam.demo.service;

import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.jam.demo.entity.User;
import com.jam.demo.mapper.UserMapper;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
import org.springframework.util.ObjectUtils;
import org.springframework.util.StringUtils;

import java.time.LocalDateTime;

/**
 * 用户服务类
 * @author ken
 */
@Slf4j
@Service
publicclass UserService extends ServiceImpl<UserMapper, User> {

    /**
     * 新增用户
     * @param user 用户信息
     * @return 新增结果（true：成功，false：失败）
     */
    public boolean addUser(User user) {
        // 参数校验
        if (ObjectUtils.isEmpty(user)) {
            log.error("新增用户失败：用户信息为空");
            returnfalse;
        }
        if (StringUtils.isEmpty(user.getUsername(), "用户名不能为空")) {
            log.error("新增用户失败：用户名不能为空");
            returnfalse;
        }

        // 设置创建时间
        user.setCreateTime(LocalDateTime.now());
        boolean saveResult = save(user);
        if (saveResult) {
            log.info("新增用户成功：用户名={}，用户ID={}", user.getUsername(), user.getId());
        } else {
            log.error("新增用户失败：用户名={}", user.getUsername());
        }
        return saveResult;
    }

    /**
     * 根据ID查询用户
     * @param id 用户ID
     * @return 用户信息
     */
    public User getUserById(Long id) {
        if (ObjectUtils.isEmpty(id)) {
            log.error("查询用户失败：用户ID为空");
            returnnull;
        }

        User user = getById(id);
        if (ObjectUtils.isEmpty(user)) {
            log.warn("查询用户不存在：用户ID={}", id);
        } else {
            log.info("查询用户成功：用户ID={}，用户名={}", id, user.getUsername());
        }
        return user;
    }
}

package com.jam.demo.controller;

import com.jam.demo.entity.User;
import com.jam.demo.service.UserService;
import io.swagger.v3.oas.annotations.Operation;
import io.swagger.v3.oas.annotations.Parameter;
import io.swagger.v3.oas.annotations.media.Content;
import io.swagger.v3.oas.annotations.media.Schema;
import io.swagger.v3.oas.annotations.responses.ApiResponse;
import io.swagger.v3.oas.annotations.tags.Tag;
import lombok.extern.slf4j.Slf4j;
import org.springframework.http.HttpStatus;
import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.*;
import org.springframework.util.ObjectUtils;

import javax.annotation.Resource;

/**
 * 用户控制器
 * @author ken
 */
@Slf4j
@RestController
@RequestMapping("/user")
@Tag(name = "用户管理接口", description = "提供用户新增、查询等接口")
publicclass UserController {

    @Resource
    private UserService userService;

    /**
     * 新增用户
     * @param user 用户信息
     * @return 新增结果
     */
    @PostMapping("/add")
    @Operation(summary = "新增用户", description = "传入用户信息（用户名、年龄、邮箱），新增用户")
    @ApiResponse(responseCode = "200", description = "新增成功", content = @Content(schema = @Schema(implementation = Boolean.class)))
    public ResponseEntity<Boolean> addUser(@RequestBody User user) {
        boolean result = userService.addUser(user);
        returnnew ResponseEntity<>(result, HttpStatus.OK);
    }

    /**
     * 根据ID查询用户
     * @param id 用户ID
     * @return 用户信息
     */
    @GetMapping("/get/{id}")
    @Operation(summary = "根据ID查询用户", description = "传入用户ID，查询用户详细信息")
    @ApiResponse(responseCode = "200", description = "查询成功", content = @Content(schema = @Schema(implementation = User.class)))
    public ResponseEntity<User> getUserById(
            @Parameter(description = "用户ID", required = true) @PathVariable Long id
    ) {
        User user = userService.getUserById(id);
        returnnew ResponseEntity<>(user, HttpStatus.OK);
    }
}

package com.jam.demo;

import com.baomidou.mybatisplus.annotation.DbType;
import com.baomidou.mybatisplus.extension.plugins.MybatisPlusInterceptor;
import com.baomidou.mybatisplus.extension.plugins.inner.PaginationInnerInterceptor;
import org.mybatis.spring.annotation.MapperScan;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.annotation.Bean;

/**
 * 应用启动类
 * @author ken
 */
@SpringBootApplication
@MapperScan("com.jam.demo.mapper")
publicclass ElkDemoAppApplication {

    public static void main(String[] args) {
        SpringApplication.run(ElkDemoAppApplication.class, args);
    }

    /**
     * MyBatis-Plus分页插件
     * @return 分页拦截器
     */
    @Bean
    public MybatisPlusInterceptor mybatisPlusInterceptor() {
        MybatisPlusInterceptor interceptor = new MybatisPlusInterceptor();
        interceptor.addInnerInterceptor(new PaginationInnerInterceptor(DbType.MYSQL));
        return interceptor;
    }
}

# 输入：读取Java应用日志（JSON格式）
input {
  file {
    path => ["/var/log/elk-demo-app/app.log"]
    start_position => "beginning"
    sincedb_path => "/home/elk/logstash/data/sincedb_app"
    stat_interval => 1
    codec => "json"  # 直接解析JSON格式日志，无需Grok
  }
}

# 过滤：数据清洗与转换
filter {
  # 时间格式标准化（将timestamp字段转为@timestamp）
  date {
    match => { "timestamp" => "yyyy-MM-dd HH:mm:ss.SSS" }
    target => "@timestamp"
    timezone => "Asia/Shanghai"
    remove_field => ["timestamp"]  # 移除原始timestamp字段
  }

  # 字段加工
  mutate {
    add_field => { "service_name" => "elk-demo-app" "env" => "prod" }
    # 异常日志添加标签
    if [level] == "ERROR" {
      add_tag => ["error_log"]
    }
  }
}

# 输出：写入ES（按天生成索引）
output {
  elasticsearch {
    hosts => ["http://localhost:9200"]
    index => "app-log-%{+YYYY.MM.dd}"
    document_id => "%{[@timestamp]}-%{thread}-%{class}"
    flush_size => 5000
    idle_flush_time => 10
    template => "/home/elk/logstash/config/app-template.json"
    template_name => "app-log-template"
    template_overwrite => true
  }

  # 调试输出（生产环境可注释）
  stdout {
    codec => rubydebug
  }
}

{
  "index_patterns": ["app-log-*"],
"settings": {
    "number_of_shards": 3,
    "number_of_replicas": 1
  },
"mappings": {
    "properties": {
      "level": { "type": "keyword" },
      "thread": { "type": "keyword" },
      "class": { "type": "keyword" },
      "message": { "type": "text", "analyzer": "ik_smart" },
      "exception": { "type": "text", "analyzer": "ik_smart" },
      "service_name": { "type": "keyword" },
      "env": { "type": "keyword" }
    }
  }
}

mvn clean package -Dmaven.test.skip=true

java -jar target/elk-demo-app-0.0.1-SNAPSHOT.jar

cd /home/elk/logstash
./bin/logstash -f config/app-log-to-es.conf

# elasticsearch.yml
cluster.name:elk-prod-cluster
node.name:node-1
path.data:/data/elasticsearch/data
path.logs:/var/log/elasticsearch
network.host:192.168.1.101
http.port:9200
# 集群节点发现
discovery.seed_hosts:["192.168.1.101","192.168.1.102","192.168.1.103"]
# 初始化主节点
cluster.initial_master_nodes:["node-1","node-2","node-3"]
# 启用安全验证（生产环境必须开启）
xpack.security.enabled:true
xpack.security.http.ssl.enabled:true
# 内存锁定（避免ES内存被交换到磁盘）
bootstrap.memory_lock:true
# 最大分片数限制
cluster.routing.allocation.total_shards_per_node:20

# jvm.options
-Xms16g  # 堆内存16GB（物理内存32GB时）
-Xmx16g
-XX:+UseG1GC  # 使用G1垃圾收集器
-XX:MaxGCPauseMillis=200  # 最大GC暂停时间200ms

# pipelines.yml
-pipeline.id:main
path.config:"/home/elk/logstash/config/*.conf"
pipeline.workers:8# 8核CPU设置8个工作线程
pipeline.batch.size:2000# 每个线程批量处理2000条
pipeline.batch.delay:5# 批量延迟5ms
queue.type:persisted# 启用持久化队列（避免宕机数据丢失）
queue.max_bytes:10gb# 队列最大容量10GB
queue.checkpoint.writes:1000# 每1000条数据 checkpoint 一次

# kibana.yml
server.port:5601
server.host:"192.168.1.104"
elasticsearch.hosts:["https://192.168.1.101:9200","https://192.168.1.102:9200","https://192.168.1.103:9200"]
# 认证配置
elasticsearch.username:"kibana_system"
elasticsearch.password:"kibana123"
# 优化会话超时时间
xpack.security.session.idleTimeout:86400000# 24小时
# 启用监控
xpack.monitoring.enabled:true
xpack.monitoring.ui.container.elasticsearch.enabled:true

# 1. 创建备份仓库
curl -X PUT "https://localhost:9200/_snapshot/elk_backup" -H "Content-Type: application/json" -u elastic:elastic123 -d '
{
  "type": "fs",
  "settings": {
    "location": "/mnt/elk_backup",
    "compress": true
  }
}'

# 2. 创建快照（备份所有索引）
curl -X PUT "https://localhost:9200/_snapshot/elk_backup/snapshot_$(date +%Y%m%d)" -u elastic:elastic123

# 3. 定时备份（通过crontab）
0 0 * * * curl -X PUT "https://localhost:9200/_snapshot/elk_backup/snapshot_$(date +%Y%m%d)" -u elastic:elastic123

tail -f /home/elk/logstash/logs/logstash-plain.log

ls -l /var/log/nginx/access.log  # 确认文件存在
sudo chmod 644 /var/log/nginx/access.log  # 赋予读取权限

curl http://localhost:9200/_cat/indices?v

cat /home/elk/kibana/config/kibana.yml | grep elasticsearch.hosts

elasticsearch.username: "kibana_system"  # Kibana默认服务用户
elasticsearch.password: "kibana123"      # 对应密码

curl -u kibana_system:kibana123 http://localhost:9200/_cluster/health

telnet 192.168.1.101 9200  # 替换为ES节点IP和端口

sudo firewall-cmd --permanent --add-port=9200/tcp
sudo firewall-cmd --reload

curl -u elastic:elastic123 http://localhost:9200/_cat/health?v

curl -X POST "http://localhost:9200/_reindex" -H "Content-Type: application/json" -d '
{
  "source": { "index": "old-index" },
  "dest": { "index": "new-index", "settings": { "number_of_shards": 3, "number_of_replicas": 1 } }
}'

{
  "query": { "match": { "service_name": "nginx" } },
  "sort": [ { "@timestamp": "asc" } ],
  "search_after": [ "2024-10-01T00:00:00Z" ],  # 上一页最后一条数据的sort值
  "size": 100
}

tail -f /var/log/filebeat/filebeat.log

# filebeat.yml 核心配置
filebeat.inputs:
- type: log
  paths:
    - /var/log/app/*.log  # 正确的日志路径
output.logstash:
  hosts: ["192.168.1.105:5044"]  # 正确的Logstash地址（5044为Filebeat输入插件默认端口）

queue.mem:
  events: 4096  # 内存队列大小
queue.disk:
  path: /var/lib/filebeat/queue  # 磁盘队列路径
  max_size: 10GB  # 磁盘队列最大容量

filebeat.inputs:
-type:log
paths:
    -/var/log/app/app.log
harvester_limit:1000
close_inactive:5m
scan_frequency:10s
tail_files:false