AWS上的EFK环境部署

AWS上的EFK环境部署

1.准备工作及组件

本章使用自建服务以及aws服务来配置使用。

2.Filebeat

2.1 配置

由于本文档讲解了多个版本的filebeat,所以不管是2.1或者2.2章节,有些配置可能会不同,但是也是做了少量更改。

2.1.1 关于配置

2.1.1.1 Processors1

在filebeat 匹配的时候,不管怎么样都会报以下类似的错误.

json/json.go:51 Error decoding JSON: json: cannot unmarshal number into Go value of type map[string]interface {}

如果想忽略这些错误,那么filebeat不能配置add_error_key,这个配置一旦配置以后, json.message_key就不能自己指定了,那么key就是默认的message。Logstash match的时候字段就是message。

所以这里有几个关系型很强的配置,这些配置是关于收集docker日志时转换成为json的作用,所以关联性很强,所以下面使用表格来表示。

总结讲解;无论是什么版本,推荐最后一组配置,这样可能会失去自定义key的方式,但是如果一旦出现报错,因为只要开启了add_error_key字段,那么一定会有error字段,在k8s field方面和其它一些展示方面功能无法实现。

2.1.1.2 processors2

7.3.1官方连接: ​​-> processor 1 -> event1 -> processor 2 -> event2 ...

简单来说,我的日志可以经过第二次的 processor的顾虑,可以是日志过滤的更详细。这里就用一个json的案例来讲解。

这种方法和processors的配置4人组没什么区别，用配置4人组和这个都可以,只是这个没有配置4人组的add_error_key的功能 #processors: #- decode_json_fields: #fields: ['log'] #target: json

2.1.1.3 k8s field字段

截止2019年9月7日 19:52:43 想要收集K8s字段必须满足一下条件.

1.K8s收集参数要打开,6.7.2之后版本都可以。

2.Filebeat必须部署在k8s集群内,且网络默认不能是宿主机模式。

3.需要对应的RBAC权限。 （该文件在代码文件夹里）

4.2019年9月8日 15:33:37 eks平台不能部署在集群外部,可能也是因为eks密钥的原因导致无法识别.

2019年9月26日 再次测试,虽然有配置在集群外的配置,但是尝试各种方式,还是无法得到k8s字段信息,可能暂时不支持部署在集群外.

在filebeat6.7.2与7.3.1里面,inputs方式不同,在7.3.1版本里面 type为docker提示将要废除,改为containers,而且路径变为/var/log/containers/,在该目录下,docker的日志是一个软连接,这个日志名包含了pod名称、命名空间和dockerid.

[root@ip-10-5-6-174 ec2-user]# cd /var/log/containers/ [root@ip-10-5-6-174 containers]# ls customer-srv-bcf745787-v7z7t_test_customer-srv-607350b200b1d0770532037a2726ceeb6d01da7225f1c0e077f5c6891790d93b.log [root@ip-10-5-6-174 containers]# ll

这个inputs的路径和收集k8s field没有关系,路径设置成哪里都无所谓,起初认为需要收集的日志名称需要有 pod 名称、命名空间等才能收集到,后面发现其实只要开启k8s部署在容器内收集就可以收集到.

但是下面的matchers.log_path 跟版本有关,7.3.1 如果input.containers 那么就需要增加.

以下收集k8s日志信息配置。

1. processors: 2. - add_kubernetes_metadata: 3. in_cluster: true 4. #这是部署在集群外部的收集方式,但现在还未实现，如果部署在集群内部就不需要增加hosts这个key 5. #host: '{NODE_NAME}' 6. #kube_config: /tmp/.kube/config 7. #使用 pod的标签 来获取pod的字段，这是api实现的，具体实现方式不详,app和name是我的deploy自定义的 8. #后面经过实测,不管哪个版本,要不要这个参数都可以9. include_labels: 10. - app 11. - k8s-app 12. - k8s-ns 13. - name 14. #可以使用pod模版的annotations信息来获取k8s信息,这里没有用到,但是配置保留 15. #include_annotations: 16. #- k8s.cloud/controller-kind 17. # 该参数是否需要取决于filebeat.input的类型,在7.3.1中,如果使用container,那么必须加上以下参数,如果使用input.docker,那么可以无视。18. matchers: 19. - logs_path: 20. logs_path: /var/log/containers

配置精简后如下:

1. processors: 2. - add_kubernetes_metadata: 3. in_cluster: true 4. include_labels: 5. - app 6. - k8s-app 7. - k8s-ns 8. - name 9. matchers: 10. - logs_path: 11. logs_path: /var/log/containers

在filebeat 6.7.2里面只需要开启以下就可以配置

processors: - add_kubernetes_metadata: in_cluster: true

2.1.2 Filebeat.yml

6.7.2和7.3.1的区别在于filebeat.inputs 的type不同以及收集k8s字段的时候有一些小区别。

包含k8s field字段获取

1. #logging.level: debug#filebeat日志等级2. filebeat.inputs: 3. #输入类型,经过测试,docker和file的收集方式区别不大,docker比较方便而以 4. - type: docker 5. #这里排除以下开头的日志文件,这里的正则和常用正则不一样,如果有需要去官网查找,比如6. #常用正则 ^ 代表开头,这里代表取反7. exclude_files: ['filebeat*|prometheus*|grafana*|node*|kube*|logstash*|zipkin*|nacos*|jenkins*|job-admin*'] 8. containers.ids: 9. - '*' 10. #多余的示例配置项,所以保留,可删除 11. #- type: log 12. #paths: 13. #- /var/lib/docker/containers/test/*.log 14. #- type: docker 15. #containers: 16. #path: "/var/lib/docker/containers" 17. #stream: "all" 18. #ids: 19. #- "*" 20. fields_under_root: true #让这个key在json格式的顶层,而不是json values里面的一个json 21. fields: #自定义字段,可以发现logstash时作为标记 22. env: test #千万不要大写,否则索引无法识别 23. spool_size: 1024 24. idle_timeout: "5s" #后台刷新超时时间,超过定义时间后强制发送,不管spool_size是否达到,默认5秒 25. backoff: "1s" #每1秒检测一次文件是否有新的一行内容需要读取 26. tail_files: false #是否从文件末尾开始读取,针对新文件 27. harvester_buffer_size: 16384 #实际读取文件时,每次读取16384字节 28. scan_frequency: "1s" #prospector扫描新文件的时间间隔,默认10秒 29. #document_type: "QA" #老版本的fields,给文档加标签的 30. json.keys_under_root: true #让字段key位于根节点,和上面fields_under_root差不多 31. #如果输出的json顶层有同名的key,那么就回覆盖默认的,比如stout默认输出有log字段,我下面的message_key定义的就覆盖默认的 32. json.overwrite_keys: true 33. #在filebeatstout输出json里面增加error的key,具体名error.message 34. #这里有个遗留问题,如果不打开这个选项,那么message_key定义的key不会出现,输出的还是默认的message 35. json.add_error_key: true 36. #message_key是用来合并多行json日志使用的,如果配置该项还需要配置multiline的设置 37. #简单来说,多行合并就靠获取整个key的里面的值以及规则是否要合并为一行 (没有这个值,可以获取到json数据,但是并不准确) 38. #如果指定,键必须位于JSON对象的顶层,且与关键的值必须是字符串,否则不会发生过滤或多行聚合 39. json.message_key: log 40. #将filebeat解码json的错误忽略掉,input使用docker模式基本每条都报错,如果不增加,传入到logstash过滤后的结果也是错的（但是不影响收集,暂时可以忽略）41. #此行配置主要是 42. json.ignore_decoding_error: true 43. #读取很多行json,碰到下面的格式就回判断是一行json,下面是docker的默认时间开头的格式 44. multiline.pattern: '^[0-9]{4}-[0-9]{2}-[0-9]{2}' 45. #定义是否否定模式 46. #true 或 false: 默认是false,匹配pattern的行合并到上一行;true,不匹配pattern的行合并到上一行 47. multiline.negate: true 48. #指定Filebeat如果将匹配行组合成事件,在之前或者之后,取决于上面所指定的negate,after 活 before,合并到上一行的末尾或者开头 49. multiline.match: after 50. #定义超时时间,如果一开始一个新的事件在超时时间内没有发现匹配,也就发送日志,默认是5s 51. multiline.timeout: 15s 52. 53. #在上面讲解过,可以使日志过滤更详细54. processors: 55. - add_kubernetes_metadata: 56. in_cluster: true 57. include_labels: 58. - app 59. - k8s-app 60. - k8s-ns 61. - name 62. 63. #output.logstash: 64. #hosts: ["10.96.30.69:5044"] 65. #output.console: 66. #pretty: true 67. output.kafka: 68. hosts: ["kafka:9092"] 69. topic: "test-log" 70. partition.round_robin: 71. reachable_only: false 72. required_acks: 1 73. compression: gzip 74. max_message_bytes: 10000000

2.2 安装

这里只讲解k8s中的安装,使用k8s安装主要分为2个步骤:

1.镜像制作,需要注意容器和宿主机的权限的一些的限制

2. k8s yml文件的编写

2.2.1 镜像制作

2.2.1.1 6.7.2

#默认的镜像是filbeat用户,如果要收集本地的日志权限很麻烦,所以改成root user root #优化容器层,所有RUN命令放在一条执行 RUN mkdir -p /tmp/.kube && cp /usr/share/zoneinfo/Asia/Tokyo /etc/localtime && \ chmod a+x filebeat && \ rm -rf /usr/share/filebeat/filebeat.yml #没用的配置,但是还是保留,如果部署在k8s集群外会有用 #ADD config /tmp/.kube/ WORKDIR /usr/share/filebeat/ #把自定义的配置放入官方镜像 COPY filebeat.yml /usr/share/filebeat/filebeat.yml #自带镜像已经有了启动,所以更新后这里也不需要了 #ENTRYPOINT /usr/share/filebeat/filebeat -e -c /usr/share/filebeat/filebeat.yml

2.2.1.2 7.3.1

7.3.1配置和6.7.2差不多,主要就是镜像版本不同而已。

FROM docker.elastic.co/beats/filebeat:7.3.1 user root #RUN mkdir /root/.kube #ADD config /root/.kube/ RUN cp /usr/share/zoneinfo/Asia/Tokyo /etc/localtime && \ rm -rf /usr/share/filebeat/filebeat.yml WORKDIR /usr/share/filebeat/ COPY filebeat.yml /usr/share/filebeat/filebeat.yml

2.2.2 K8s配置文件

由于服务都是通用模版,所以这里仅贴出数据映射到宿主机的配置,具体的其它配置可参考模版文件。

2.2.2.1 6.7.2

1. volumeMounts: 2. #filebeat的offset位置,用这里记录到日志收集到哪里,当filebeat宕掉的时候,可以再次从node 本机读取记录 3. - name: filebeat-offset 4. mountPath: /usr/share/filebeat/data 5. readOnly: false 6. #docker容器日志的位置,需要从容器映射到宿主机 7. - name: docker-dir 8. mountPath: /var/lib/docker/containers 9. #这个只给只读权限 10. readOnly: true 11. #docker的sock文件位置,可以使用它执行docker命令,filebeat容器使用不到,但是配置保留 12. - name: docker-sock 13. mountPath: /var/run/docker.sock 14. readOnly: true 15. #filebeat的配置文件,使用configmap来更新配置,更新之后filebeat要滚动更新一次 16. #该流程可以优化,可以将配置打入docker镜像中.这是本人为了测试configmap的方式 17. - name: filebeat-config 18. mountPath: /usr/share/filebeat/conf 19. #将本地时间挂载到pod内20. - name: localtime21. mountPath: /etc/localtime22. #与宿主机共同共享一个网络ip,这样可以避免多分配一个ip给该实例 23. #如果想获取到K8s field字段,千万不要开这2个选项24. #hostNetwork: true 25. #hostPID: true 26. #由于设置了hostNetwork,dns的解析方式发生了改变,这个设置项,默认4个选项,默认是集群内dns解析 27. #由于设置了hostNetwork,所以变成了宿主机dns,设置这个选项,可以是宿主机dns和集群内部dns共存 28. #dnsPolicy: ClusterFirstWithHostNet 29. terminationGracePeriodSeconds: 30 30. volumes: 31. - name: filebeat-offset 32. hostPath: 33. path: /opt/filebeat 34. type: DirectoryOrCreate 35. - name: docker-dir 36. hostPath: 37. path: /var/lib/docker/containers 38. - name: docker-sock 39. hostPath: 40. path: /var/run/docker.sock 41. - name: filebeat-config 42. configMap: 43. name: filebeat-config

2.2.2.2 7.3.1

7.3.1 和6.7.2区别在于,多映射了一个/var/log/container的文件夹,这个是具有pod信息表示的日志文件,并且7.3.1取消了configmap。

1. volumeMounts: 2. - name: filebeat-offset 3. mountPath: /usr/share/filebeat/data 4. readOnly: false 5. - name: docker-dir 6. mountPath: /var/lib/docker/containers 7. readOnly: true 8. #和6.7.2的区别在于多映射了一个这样的文件夹 9. - name: varlog 10. mountPath: /var/log 11. readOnly: true 12. - name: docker-sock 13. mountPath: /var/run/docker.sock 14. readOnly: true 15. #模式要调试成为非host模式,让filebeat运行在集群内,和6.7.2一样关闭 16. #hostNetwork: true 17. #hostPID: true 18. #dnsPolicy: ClusterFirstWithHostNet 19. terminationGracePeriodSeconds: 30 20. volumes: 21. - name: filebeat-offset 22. hostPath: 23. path: /opt/filebeat 24. type: DirectoryOrCreate 25. - name: docker-dir 26. hostPath: 27. path: /var/lib/docker/containers 28. - name: varlog 29. hostPath: 30. path: /var/log 31. - name: docker-sock 32. hostPath: 33. path: /var/run/docker.sock

3.Logstash

由于logstash 6.7.2和7.3.1唯一不同的就是镜像版本不同,其余的完全一样,所以这里不列多余的篇章。

3.1 配置

1. input { 2. #屏幕输入,用于测试是否正常 3. #stdin { 4. #} 5. #filebeats等输入 6. #beats { 7. #port => 5044 8. #} 9. #kafka常规操作 10. kafka { 11. bootstrap_servers => "10.1.50.152:9092,10.1.51.209:9092,10.1.51.236:9092" 12. #启用多实例消费,可以部署第二个logstash. 13. #所以要设置 1.相同的topic; 2.相同的groupid; 3 不同的client(这个最重要,根据kafka的原理使用) 14. group_id => "logstash_log_group" 15. #如果要启用第二个logstash,那么这里的客户端id改成2 16. client_id => "logstash_log_group_1" 17. #这个参数就是控制如果有多个logstash,那么决定要不要让其他logstash来处理kafka消息 18. #该参数默认为1,如果这个参数大于kafka topic 的分区数,那么整个topic的所有消息都由这个logstash来处理 19. #简单来说,topic分区比这个值大才会由其它logstash处理。 20. #consumer_threads => 10 21. #bootstrap_servers => "192.168.1.120:9092,192.168.1.201:9092,192.168.1.233:9092" 22. topics => "ap-lcm-prod-log" 23. codec => "json" 24. auto_offset_reset => "latest" 25. } 26. } 27. 28. filter { 29. #if [env] == "test" { 30. #grok千万不要有多个,如果有多个,会造成即使匹配成功,还是会出现"tags" => [ [0] "_grokparsefailure"], 这个报错,所以最好用下面的第二种写法 31. grok { 32. #这里是match的一种书写方式,字典类型,如果使用列表那么就是['log','part...'] 33. #match => { "log" => "(%{TIMESTAMP_ISO8601:times})\s*(?test|sand|prod)\s*(?[a-zA-Z]{1,5})\s*\[(?[a-zA-Z0-9\-\_ ]*),(?[a-zA-Z0-9\-\_ ]*),(?[a-zA-Z0-9\-\_ ]*),(?[a-zA-Z0-9\-\_ ]*)\]\s*(?.*)" } 34. #使用列表,可以定多种的match匹配规则,比如match nginx和match 业务的规则,如果需要有多种规则,就需要使用下面的方式 35. #这里的log就是filebeat发来的一个json里面的一个key,可能key是message或者其它自定义字段 36. #第一个是默认自带的时间戳格式, \s*表示零个或者多个空格, (?)代表logstash输出的标签,后面是匹配的正则表达式 37. #正则表达只有多联系,这里不做解释 38. match => [ "message" , "(%{TIMESTAMP_ISO8601:times})\s*(?test|sand|prod)\s*(?[a-zA-Z]{1,5})\s*\[(?[a-zA-Z0-9\-\_ ]*),(?[a-zA-Z0-9\-\_ ]*),(?[a-zA-Z0-9\-\_ ]*),(?[a-zA-Z0-9\-\_ ]*)\]\s*(?.*)", 39. "message" , "(%{TIMESTAMP_ISO8601:times})\|(?nginx|nginx\-public)\|(?test|sand|prod)\|(?[a-zA-Z0-9\.\/\-\:\?\=\&]+|-)\|(%{IPV4:remote_addr}|-)\|(%{USERNAME:user}|-)\|(%{IPV4:realip}|-)\|(%{WORD:request_method}) (?[a-zA-Z0-9\.\/\-\:\?\=\& ]+)\|(%{URIHOST:]+|-)\|(?([0-9.]{0,3}[, ]{0,2})+|-)\|(%{URIHOST:upstream_addr}|-)\|(%{NUMBER:upsteam_status}|-)\|(?([0-9.]{0,3}[, ]{0,2})+|-)\|(?.*)\|" ] 40. #logstash输出的json格式里删除掉这些key 41. remove_field => "beat" 42. remove_field => "tags" 43. remove_field => "log" 44. remove_field => "input" 45. remove_field => "prospector" 46. remove_field => "stream" 47. remove_field => "time" 48. remove_field => "error" 49. } 50. 51. #grok { 52. #match => { "source" => "\/var\/lib\/docker\/containers\/[a-zA-Z0-9]*\/(?.*)" } 53. #remove_field => "source" 54. #} 55. 56. #mutate过滤器能够帮助你修改指定字段的内容 57. mutate { 58. #大小写转换 59. lowercase => [ "server_name" ] 60. #数组类型,没有默认设置,该参数设置只针对string类型，如果不是string类型的，什么也不做,由于logstash对_敏感不识别,所以转换成- 61. gsub => [ 62. "server_name", "_", "-" 63. ] 64. } 65. 66. mutate { 67. #nginx转换,将数据类型转换为int,只有int才能绘图,为绘图准备 68. convert => { 69. #下面都是nginx通过正则过滤出来的字段 70. "status" => "integer" 71. "send" => "integer" 72. "request_time" => "float" 73. "upstraem_status" => "integer" 74. "upstraem_time" => "float" 75. } 76. } 77. 78. #将日志的时间覆盖成为timestsmp,就是把应用程序输出的时间覆盖logstash的时间戳,做到时间戳是日志记录的时间 79. date { 80. match => ["times", "yyyy-MM-dd HH:mm:ss.SSS", "ISO8601"] 81. #match => ["times", "yyyy-MM-dd HH:mm:ss.SSS"] 82. target => "@timestamp" 83. timezone => "Japan" 84. } 85. #} 86. } 87. 88. output { 89. if "_grokparsefailure" not in [tags] { 90. #stdout { 91. #codec=>rubydebug 92. #} 93. elasticsearch { 94. #user => 'elastic' password => 'huawei' 95. #hosts => ["10.5.2.39:9200"] 96. #这些标签变量都是收到的json里面提取的 97. hosts => [" 98. #下面都是nginx通过正则过滤出来的字段 99. ssl => true 100. #自建无法使用contain_name来标识分类索引,这点比较可惜,有时候server_name在容器报错情况下无法无法获取server_name 101. index => "%{branch}-%{server_name}-%{+YYYY.MM.dd}" #field不能大写,否则无法识别 102. } 103. } 104. }

3.2 安装

3.2.1 Logstash 容器安装

由于logstash在需要自定义配置文件,所以,本篇章讲解,docker运行logstash需要那些注意事项。

1.创建相关配置文件

logstash.yml

空文件或者从tgz包里面拷贝一个当前版本默认的

pipelines.yml(那个小杠杠很重要)

读取该目录下的*.conf文件作为主配置文件

- pipeline.id: main path.config: "/usr/share/logstash/config/*.conf"

还有以下文件,都可以从默认的tgz包中去拷贝使用,有一点需要注意,如果核心配置文件的后缀为.yml,那么logstash.yml这个文件建议删除,不然logstash会无法工作。

jvm.options lcmv2.conf lcmv2.conf.1 log4j2.properties logstash.yml pipelines.yml startup.options

3.2.2 镜像制作

FROM lizexiong/logstash:6.4.2 user root #RUN cp /usr/share/zoneinfo/Asia/Shanghai /etc/localtime RUN cp /usr/share/zoneinfo/Asia/Tokyo /etc/localtime #删除默认镜像里面的配置 RUN rm -rf /usr/share/logstash/config/* #把自己准备的配置文件拷贝过去 COPY config/* /usr/share/logstash/config/ EXPOSE 5044 EXPOSE 9600

3.2.3 Logstash k8s配置

K8s配置文件没有什么需要注意的

4.Cognito

Amazon Cognito 提供用户池和身份池。用户池是为您的应用程序提供注册和登录选项的用户目录。身份池提供 AWS 凭证以向用户授予对其他 AWS 服务的访问权限。

在创建es服务之前来创建cognito是因为创建es服务时,需要制定用户池和身份池。

因为在创建es之后会自动对用户池做一些设置,所以这里仅贴出需要设置的项。

4.1 创建用户池

2.勾选予许name和email登录

3.勾选仅予许管理员创建.

4.创建应用程序客户端,在创建身份池的时候需要关联

5.给用户池添加域名

4.2 创建身份池

2.设置身份池权限

在开始使用新的Amazon Cognito身份池之前，必须分配一个或多个IAM角色，以确定应用程序最终用户对AWS资源的访问级别。标识池定义两种类型的标识:经过身份验证的和未经身份验证的。每个人都可以在IAM中分配自己的角色。经过身份验证的身份属于经过公共登录提供者(Amazon Cognito用户池、Facebook、谷歌、SAML或任何OpenID连接提供者)或开发人员提供者(您自己的后端身份验证过程)身份验证的用户，而未经身份验证的身份通常属于来宾用户。

当Amazon Cognito接收到用户请求时，服务将确定该请求是经过身份验证的还是未经身份验证的，确定哪个角色与该身份验证类型相关联，然后使用附加到该角色的策略来响应该请求。

3.创建完成,可以通过编辑身份池来查看详细信息和修改信息

5.1 Elastaticsearch

5.1.1 aws service

1.创建aws es服务,配置选项

生产环境开启专用主实例

设置存储大小和是否加密以及自动快照的时间,这里全部选择默认

2.设置权限

这里一定要注意,vpc访问和公有访问权限的区别.

vpc访问:是只有这个vpc下的网段才能访问该es和kibana,设置该选项,那么只能vpc到该vpc内网进行访问.公有访问权限:可设置策略,也可以设置对ip访问.

所以下面选择公有访问权限,并启动cognito认证

3.访问策略

在下一步之前,这里首先随便设置一个策略让整个集群创建流程走完

4.现在讲解最重要的策略设置

刚才第三步设置的策略可以忽略,因为以下我们要针对几个地方做授权。

Cognito的用户访问kibana的权限.Logstash访问es的权限,logstash访问es不管是不是在同一vpc下,那么都会走外网,所以要把logstash的外网ip给放行.

番外篇

为了防止后期找不到要放行的资源在哪,这里明确指出在哪里需找这些消息

Kibana访问角色的资源。

身份池IAM角色会显示在这里

Resource资源

这个创建一个默认模版就可以得到resource资源信息

还是将模版配置贴出来

{ "Version": "2012-10-17", "Statement": [ { "Effect": "Allow", "Principal": { "AWS": "arn:aws:iam::858659433780:role/Cognito_ap_lcm_prod_esIdentityPoolAuth_Role" }, "Action": "es:*", "Resource": "arn:aws:es:ap-northeast-1:858659433780:domain/ap-lcm-prod-es/*" }, { "Effect": "Allow", "Principal": { "AWS": "*" }, "Action": "es:*", "Resource": "*", "Condition": { "IpAddress": { "aws:SourceIp": [ "3.112.77.138/32", "13.230.137.66/32" ] } } } ]}

5.1.2 自建

自建es本章节保留

5.2 Kibana

由于kibana章节较为简单,创建es服务后自带kibana服务,所以这里只是简单演示一下使用,没有安装等讲解。

自建kibana参考本人elk部署博客。

5.2.1 创建用户

1.在congito给kibana创建登录用户

5.2.2 Kibana时间戳

在创建索引后之前,可以先尝试把时间戳的时间更改至你想看到的时间。比如,在kibana的时候有三个时间字段展示,这里拿nginx做演示。

Time时间是kibana自带的,无法更改这个字段。

Timestamp是在logstash里面自己去生成的.

Times我们自己通过logstash字段过滤出来的自定义时间格式字段.

有的时候kibana暂时的时间跟我们的logstash过滤的时间明显不一样,所以这里就是kibana的展示问题,因为kibana的时间是默认根据浏览器来的,如果服务器在东京,浏览器在中国,那么会看到,明明是东京的时间,却显示的还是中国的时间,那么就需要调整kibana的设置的。

1.进入 kibana设置→ managerment → Advanced Settings

2.选择与服务器相同的时区,问题得到解决

这里有一个遗留问题,times是自定义的时间,但是不知道为什么,在logstash里面显示正常,在kibana就变成了utc的时间。

问题解决:2019年8月27日 23:22:25

Times自定义时间不一致,显示utc,那是因为在logstash的 grok里面,没有用数组的方式写2种规则的匹配,导致即使匹配到了日志,依然还是会将出现"tags" => [ [0] "_grokparsefailure"],这样,times的日志就变成了utc,所以,一定要将logstash的输出保持正确。

5.2.3 新建索引及保存查询结果

1.创建索引

Index patterns→create index pattern,选择需要建立的索引

由于在logstash的时候把日志输出的时间覆盖了timestamp,所以这里要选择使用自己的时间戳。

2.保存查询结果

由于有些查看方式不太相同,切换一个索引后,整个字段表现的非常混乱,所以主要是一下两个按钮。

Save之后的结果可以通过open打开

5.2.4 创建 visualizations

首先创建nginx索引,创建可视化使用nginx来演示.

5.2.4.1 metris

选择索引

比如nginx日志的总条目数,请求平均时间,请求最大时间等等。

Nginx日志总条目数,使用count

然后选择max,这里会让我们选择字段,然后也可以给这个字段添加一个标签

最小时间也是如此。

5.2.3.2 Pie

选择索引

这里是饼图的默认选择,和metris一样,选择字段

5.2.5 创建dashboard

这里仪表板,就是把visualizations聚合显示在一起。

然后选择add添加要一起展示的图形

然后按照步骤,将有用的图形全部展示在一起。

5.2.6 创建line

由于暂时没有这样的需求,本章保留

5.3 MSK

5.3.1 创建MSK

选择名称→ 选择vpc和选择kafka版本

2.选择可用区,选择3个可用区

因为3台成集群,以下子网是本项目中规划的3个pub的网络

3.选择配置，这里我们首先默认,集群创建完毕后更改

4.创建broker的数量,这里创建1个,3个可用区就是3个

5.给集群打上一个标签,这步骤可忽略

6.选择存储大小

这里选择500G,可动态扩容

7.接下来比较重要,选择是否加密

Kafka集群间传输加密,这里选择不开启;

客户机与brokers这里选择予许明文和不加密;

静止数据加密这里我们让AWS去管理;

8.客户端连接身份验证

是否选择客户端到集群使用tls,这里不选择。

9.监控

收费监控,监控项最多,全部选择

10.自定义设置选择安全组

11.查看kafka的连接信息

现在可以进入集群查看kafka的地址等信息

12.Kafka更改配置官方连接

​​　　class="data-table" data-id="t7a7e9d1-FMVGEH8m" data-transient-attributes="class" data-width="802.014px" style="width: 100%; outline: none; border-collapse: collapse;">

以下已全部开启多行匹配以及keys_under_root(这个参数设置为ture,后期不需要做多的测试)，只要开启了add_error，有了报错,就不会获取k8s field字段

作者：​​小家电维修​​

转世燕还故榻,为你衔来二月的花。

版权声明：本文内容由网络用户投稿，版权归原作者所有，本站不拥有其著作权，亦不承担相应法律责任。如果您发现本站中有涉嫌抄袭或描述失实的内容，请联系我们jiasou666@gmail.com 处理，核实后本网站将在24小时内删除侵权内容。