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

服务

版本

作用

filebeat

6.7.2→ 7.3.1

节点日志收集,只完成少量比如多行合并工作

logstash

6.4.2→7.3.1

日志过滤,最重工作的日志过滤在这里

Elastaticsearch

7.1

aws全托管服务

Kibana

7.1.1

aws全托管服务,由于全托管,无法使用插件

认证

无

cognito,aws托管服务,完成kibana认证工作

消息队列

2.2.1

MSK,aws全托管kafka。（本章节不做讲解，详细请查阅系统aws指导文档），本人自建单机kafka做的文档记录

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的作用,所以关联性很强,所以下面使用表格来表示。

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

keys_under_root

overwrite_keys

add_error_key

message_key

结果

TRUE

default

default

default

开启多行匹配配置报错

TRUE

TRUE

default

default

开启多行匹配配置报错

TRUE

TRUE

TRUE

default

开启多行匹配配置报错

TRUE

TRUE

TRUE

TRUE

1.默认消息key自定义为log,k8s field无法获取

2.默认消息key自定义为非log,k8s field获取正常

3.出现error.message

TRUE

default

TRUE

TRUE

1. 默认消息key自定义为log,默认的message、logs字段被覆盖,没有主message keys,也就是说没有任何匹配的我们需要拿到的消息,k8s field无法获取

2. 默认消息key自定义为非log,主消息就是这个自定义的非log,k8s field获取正常.

3.出现error.message

TRUE

default

default

TRUE

1.关闭add_error_key,message_key无论设置成什么,主message key值都是message,k8s field获取正常.

2. 没有error.message

TRUE

TRUE

default

TRUE

1.关闭add_error_key,message_key无论设置成什么,主message key值都是message,k8s field获取正常.

2.没有error.message

3.和上面一条结果一样,开启关闭overwrite_keys在这两种配置没有区别

这里的遗留问题,就是keys_under_root 不管设置成什么,多行匹配也会去匹配不是message_key指定的值，这个疑问可能是filebeat会自动去寻找相同的键的值来匹配多行的规则

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

2.1.1.2 processors2

　　7.3.1官方连接: https://www.elastic.co/guide/en/beats/filebeat/7.3/decode-json-fields.html

　　event -> 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
lrwxrwxrwx 1 root root 69 Sep 6 04:15 customer-srv-bcf745787-v7z7t_test_customer-srv-607350b200b1d0770532037a2726ceeb6d01da7225f1c0e077f5c6891790d93b.log -> /var/log/pods/d0a90717-d05c-11e9-b7c7-0a2ce92fd402/customer-srv/1.log

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

2.2 安装

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

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

　　　　2. k8s yml文件的编写

2.2.1 镜像制作

2.2.1.1 6.7.2

FROM docker.elastic.co/beats/filebeat: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: localtime
21. mountPath: /etc/localtime
22. #与宿主机共同共享一个网络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

    　　由于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 {
28. #if [env] == "test" {
29. #grok千万不要有多个,如果有多个,会造成即使匹配成功,还是会出现"tags" => [ [0] "_grokparsefailure"], 这个报错,所以最好用下面的第二种写法
30. grok {
31. #这里是match的一种书写方式,字典类型,如果使用列表那么就是['log','part…']
32. #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*(?.*)" }
33. #使用列表,可以定多种的match匹配规则,比如match nginx和match 业务的规则,如果需要有多种规则,就需要使用下面的方式
34. #这里的log就是filebeat发来的一个json里面的一个key,可能key是message或者其它自定义字段
35. #第一个是默认自带的时间戳格式, \s*表示零个或者多个空格, (?)代表logstash输出的标签,后面是匹配的正则表达式
36. #正则表达只有多联系,这里不做解释
37. 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*(?.*)",
38. "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:http_host}|-)\|(%{NUMBER:status}|-)\|(%{NUMBER:send}|-)\|(?[a-zA-Z0-9\.\/\-\:\?\=\&]+|-)\|(?[a-zA-Z0-9\/\.\_\(\)\;\,\-\:\#\%\^\&\*\!\>\{\} ]+|-)\|(?([0-9.]{0,3}[, ]{0,2})+|-)\|(%{URIHOST:upstream_addr}|-)\|(%{NUMBER:upsteam_status}|-)\|(?([0-9.]{0,3}[, ]{0,2})+|-)\|(?.*)\|" ]
39. #logstash输出的json格式里删除掉这些key
40. remove_field => "beat"
41. remove_field => "tags"
42. remove_field => "log"
43. remove_field => "input"
44. remove_field => "prospector"
45. remove_field => "stream"
46. remove_field => "time"
47. remove_field => "error"
48. }
49. 51. #grok {
50. #match => { "source" => "\/var\/lib\/docker\/containers\/[a-zA-Z0-9]*\/(?.*)" }
51. #remove_field => "source"
52. #}
53. 56. #mutate过滤器能够帮助你修改指定字段的内容
54. mutate {
55. #大小写转换
56. lowercase => [ "server_name" ]
57. #数组类型,没有默认设置,该参数设置只针对string类型，如果不是string类型的，什么也不做,由于logstash对_敏感不识别,所以转换成-
58. gsub => [
59. "server_name", "_", "-"
60. ]
61. }
62. 66. mutate {
63. #nginx转换,将数据类型转换为int,只有int才能绘图,为绘图准备
64. convert => {
65. #下面都是nginx通过正则过滤出来的字段
66. "status" => "integer"
67. "send" => "integer"
68. "request_time" => "float"
69. "upstraem_status" => "integer"
70. "upstraem_time" => "float"
71. }
72. }
73. 78. #将日志的时间覆盖成为timestsmp,就是把应用程序输出的时间覆盖logstash的时间戳,做到时间戳是日志记录的时间
74. date {
75. match => ["times", "yyyy-MM-dd HH:mm:ss.SSS", "ISO8601"]
76. #match => ["times", "yyyy-MM-dd HH:mm:ss.SSS"]
77. target => "@timestamp"
78. timezone => "Japan"
79. }
80. #}
81. }
82. 88. output {
83. if "_grokparsefailure" not in [tags] {
84. #stdout {
85. #codec=>rubydebug
86. #}
87. elasticsearch {
88. #user => 'elastic' password => 'huawei'
89. #hosts => ["10.5.2.39:9200"]
90. #这些标签变量都是收到的json里面提取的
91. hosts => ["https://search-ap-lcm-prod-es-jjtulbh76rqty7neqvt5kyjywu.ap-northeast-1.es.amazonaws.com:443"]
92. #下面都是nginx通过正则过滤出来的字段
93. ssl => true
94. #自建无法使用contain_name来标识分类索引,这点比较可惜,有时候server_name在容器报错情况下无法无法获取server_name
95. index => "%{branch}-%{server_name}-%{+YYYY.MM.dd}" #field不能大写,否则无法识别
96. }
97. }
98. }

3.2 安装

3.2.1 Logstash 容器安装

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

　　原文链接: https://blog.csdn.net/qq_33547169/article/details/86629261

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配置文件没有什么需要注意的

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

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

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

4.1 创建用户池

1.点击创建用户池,输入用户池名称

2.勾选予许name和email登录

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

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

5.给用户池添加域名

4.2 创建身份池

1.点击跳转至创建身份池,输入身份池名称,配置用户池id及应用程序客户端id

2.设置身份池权限

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

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

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

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本章节保留

　　由于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.创建索引

　　点击management

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

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

2.保存查询结果

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

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

5.2.4 创建 visualizations

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

5.2.4.1 metris

1.打开visualizations,点击metris

　　选择索引

2.点击要展示的值

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

　　Nginx日志总条目数,使用count

　　Nginx请求的最大时间,那么点击add metrics

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

　　如果需要立即生效,点击右上角的箭头。

　　最小时间也是如此。

5.2.3.2 Pie

1.打开visualizations,点击pie

　　选择索引

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

如果需要立即生效,点击右上角的箭头。

　　这里还有一步，点击 “options”选择显示百分比。

5.2.5 创建dashboard

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

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

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

5.2.6 创建line

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

5.3.1 创建MSK

1. 选择名称→ 选择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更改配置官方连接

　　https://docs.aws.amazon.com/msk/latest/developerguide/msk-configuration-operations.html

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