ELK開源實時日志分析平臺，ELK是Elasticsearch,Logstash,Kibana 的縮寫，

Elasticsearch：是個開源分布式搜索引擎,簡稱ES
Logstash：是一個完全開源的工具，可以對日志進行收集，過濾，存盤到ES
Kibana: 也是一個開源和免費的工具,這里主要用作ES的可視化界面工具，用于查看日志，

環境:centos7.9

一、搭建ES

先要調高jvm執行緒數限制，修改sysctl.conf

vim /etc/sysctl.conf

修改max_map_count調大，如果沒有這個設定，則新增一行

vm.max_map_count=262144

改完保存后，執行下面命令讓sysctl.conf檔案生效

sysctl –p

拉取es鏡像

#拉取鏡像，指定版本號
 docker pull elasticsearch:7.13.2

新建elasticsearch.yml組態檔并上傳到主機目錄用于組態檔掛載，方便后面修改，這里上傳到/home/es目錄，

http.host: 0.0.0.0
#跨域
http.cors.enabled:  true 
http.cors.allow-origin:  "*"

啟動es

docker run -d -p 9200:9200 -p 9300:9300 --name es -e ES_JAVA_OPTS="-Xms128m -Xmx256m" -v /home/es/elasticsearch.yml:/usr/share/elasticsearch/config/elasticsearch.yml elasticsearch:7.13.2
#9200是對外埠，9300是es內部通信埠
#-e ES_JAVA_OPTS="-Xms128m -Xmx256m" 限制記憶體,最小128,最大256
#-v 把組態檔掛載到es的docker里的組態檔

在瀏覽器打開,ip:9200看到es資訊，就成功了，啟動可能需要一點時間，要等一會，

二、搭建Kibana

啟動kibana，這里沒有先拉取鏡像，docker沒有鏡像會到公共庫嘗試拉取，下面的一樣，

#如果沒有鏡像會自動到公共庫拉取再啟動，-e 環境配置指向es的地址
docker run -p 5601:5601 -d --name kibana -e ELASTICSEARCH_URL=http://172.16.2.84:9200 -e ELASTICSEARCH_HOSTS=http://172.16.2.84:9200 kibana:7.13.2

在瀏覽器輸入ip:5601，能看到kibana的資訊，說明成功了

三、搭建kafka

1)kafka前置需要先安裝zookeeper

#-v /etc/localtime:/etc/localtime把本機的時間掛載進docker，讓docker同步主機的時間
docker run -d --name zookeeper -p 2181:2181 -v /etc/localtime:/etc/localtime zookeeper

2)安裝kafka

docker run  -d --name kafka -p 9092:9092 -e KAFKA_BROKER_ID=0 -e KAFKA_ZOOKEEPER_CONNECT=172.16.2.84:2181 -e KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://172.16.2.84:9092 -e KAFKA_LISTENERS=PLAINTEXT://0.0.0.0:9092 -t wurstmeister/kafka
#KAFKA_BROKER_ID:kafka節點Id,集群時要指定
#KAFKA_ZOOKEEPER_CONNECT：配置zookeeper管理kafka的路徑,內網ip
#KAFKA_ADVERTISED_LISTENERS：如果是外網的，則外網ip，把kafka的地址埠注冊給zookeeper,將告訴 zookeeper 自己的地址為 XXXX，當消費者向 zookeeper 詢問 kafka 的地址時，將會回傳該地址
#KAFKA_LISTENERS： 配置kafka的監聽埠

kafka界面查看可以用 kafka tool 地址：https://www.kafkatool.com/download.html

四、搭建Logstash

新建檔案logstash.yml，logstash配置資訊

http.host: "0.0.0.0"
xpack.monitoring.elasticsearch.hosts: [ "http://172.16.2.84:9200" ]
#host多節點用逗號隔開

新建檔案logstashkafka.conf，讀取kafka資訊和寫到es的資訊

input {
     kafka {
      topics => "logkafka" #訂閱kafka的topics
      bootstrap_servers => "172.16.2.84:9092"  # 從kafka的leader主機上提取快取
      codec => "json"  # 在提取kafka主機的日志時，需要寫成json格式
            }
}
output {
  elasticsearch {
    hosts => ["172.16.2.84:9200"]
    index => "logkafka%{+yyyy.MM.dd}" #采集到es的索引名稱
    #user => "elastic"
    #password => "changeme"
  }
}

把這兩個檔案放到主機目錄上，這里放到/home/logstash

啟動logstash容器

docker run --rm -it --privileged=true -p 9600:9600 -d --name logstash -v /home/logstash/logstashkafka.conf:/usr/share/logstash/pipeline/logstash.conf -v /home/logstash/logstash.yml:/usr/share/logstash/config/logstash.yml logstash:7.13.2
#-v 把logstashkafka.conf和logstash.yml掛載到logstash容器

好了，到這里所有需要的容器都建好了運行docker ps -a看下所有容器

五、.Net Core寫日志到kafka

.Net Core只要把日志寫到kafka，然后logstash訂閱kafka把日志寫到es即可，這里是.Net Core5.0演示，

NuGet包安裝Confluent.Kafka

 class Program
    {
        static async Task Main(string[] args)
        {
            Console.WriteLine("Hello World!");
            //kafka節點
            string brokerList = "172.16.2.84:9092";
            string topicName = "logkafka";
            while (true)
            {
                Console.WriteLine($"{Environment.NewLine}請輸入發送的內容,發送topics名:{topicName}");
                string content = Console.ReadLine();
                await ConfluentKafka.PublishAsync(brokerList, topicName, content);
            }
        }
    }

 public class ConfluentKafka
    {
        public static async Task PublishAsync(string brokerList, string topicName, string content)
        {
            //生產者配置
            var config = new ProducerConfig
            {
                BootstrapServers = brokerList, //kafka節點
                Acks = Acks.None //ack機制，0不等服務器確認，1主節點確認回傳ack,-1全部節點同步完回傳ack
            };
            using (var producer = new ProducerBuilder<string, string>(config)
            .Build())
            {
                try
                {
                    //key要給值，根據key做負載均衡，不然如果多節點，key不給值會全部寫有一個磁區
                    var deliveryReport = await producer.
                        ProduceAsync(
                        topicName, new Message<string, string> { Key = (new Random().Next(1, 10)).ToString(), Value =https://www.cnblogs.com/wei325/p/ content });
                    Console.WriteLine($"向kafka發送了資料: {deliveryReport.TopicPartitionOffset}");
                }
                catch (ProduceException<string, string> e)
                {
                    Console.WriteLine($"向kafka發送資料失敗:{e.Message}");
                }
            }
        }
    }

運行程式，向kafka寫資料

查看kafka佇列，已經有資料

六、ELK日志查看

打開Kibana界面，ip:5601,打開左邊的面板

點擊索引匹配

輸入匹配符，匹配到es日志索引，下一步

索引匹配創建完畢，回傳到面板

七、其他方式寫日志

上面介紹的是kafka方式寫日志，也是ELK最佳方式，能承載大量的日志傳輸，這里列一下其他常用的方式，主要是修改logstash采集配置就可以了，

1.讀目錄檔案方式，根據給的目錄，檔案型別去讀取寫到es,logstash配置

# Sample Logstash configuration for creating a simple
# Beats -> Logstash -> Elasticsearch pipeline.

input {
 file {
 path => "D:/Log/Application/*log.log" #讀取目錄下log.log結尾的檔案
   start_position => beginning
    }
 file {
 path => "D:/Log/Application2/*log.log" #讀取目錄下log.log結尾的檔案
   start_position => beginning
    }
}
output {
  elasticsearch {
    hosts => ["172.16.2.84:9200"]
    index => "filelog"
    #user => "elastic"
    #password => "changeme"
  }
}

2.Tcp方式，通過tcp方式寫日志，在log4net和nlog中直接可配tcp請求logstash,配置

input {
 tcp{
 port => 8001
 type => "TcpLog"
 }
}
output {
  elasticsearch {
    hosts => ["172.16.2.84:9200"]
    index => "tcplog"
    #user => "elastic"
    #password => "changeme"
  }
}

3.reids，把日志推送到一個list型別的Key,logstash會訂閱這個key讀取訊息寫到es

input {
    redis {
        codec => plain
        host => "127.0.0.1"
        port => 6379
        data_type => list
        key => "listlog"
        db => 0
    }
}
output {
  elasticsearch {
    hosts => ["172.16.2.84:9200"]
    index => "redislog"
    #user => "elastic"
    #password => "changeme"
  }
}

4.RabbitMQ，把日志寫到一個佇列，讓logstash訂閱佇列獲取日志寫到es

input {
    rabbitmq {
        host => "127.0.0.1"    #RabbitMQ-IP地址
        vhost => "test"      #虛擬主機
        port => 5672             #埠號
        user => "admin"            #用戶名
        password => "123456"        #密碼
        queue => "LogQueue"      #佇列
        durable => false         #持久化跟佇列配置一致
        codec => "plain"         #格式
    }
}

output {
    elasticsearch {
        hosts => ["172.16.2.84:9200"]
        index => "rabbitmqlog"
    }
}

轉載請註明出處，本文鏈接：https://www.uj5u.com/net/295776.html

標籤：.NET Core

上一篇：dotnet 在 WPF 里顯示數學 π 的顏色

下一篇：Asp.Net Core Razor頁面中使用echarts展示圖形