CDPzを利用したz/OS-ELK連携 - (2)SYSLOG取得編

# はじめに

ここでは、SYSLOGをElasticsearchに取り込むってのをやってみます。

## 関連記事
[CDPzを利用したz/OS-ELK連携 - (1)セットアップ編](https://qiita.com/tomotagwork/private/f4fbb6d3e8b6ad084b34)
[CDPzを利用したz/OS-ELK連携 - (2)SYSLOG取得編](https://qiita.com/tomotagwork/private/5cd1d97dadaaf999c9cb)
[CDPzを利用したz/OS-ELK連携 - (3)SMF-バッチ編](https://qiita.com/tomotagwork/private/629c8bbbc95f9733e297)
[CDPzを利用したz/OS-ELK連携 - (4)SMF-リアルタイム編](https://qiita.com/tomotagwork/private/b339779769f7a5092b5a)
[CDPzを利用したz/OS-ELK連携 - (5)カスタマイズ-共通](https://qiita.com/tomotagwork/private/fe76619e8880956aa0e0)
[CDPzを利用したz/OS-ELK連携 - (6)カスタマイズ-SYSLOG](https://qiita.com/tomotagwork/private/b8e9fad37730703ef1a7)
[CDPzを利用したz/OS-ELK連携 - (7)カスタマイズ-CICS](https://qiita.com/tomotagwork/private/38091c17da9b13a4ee45)
[CDPzを利用したz/OS-ELK連携 - (8)カスタマイズ-RMF Monitor III](https://qiita.com/tomotagwork/private/7be3c192e5e1c1d7e3bc)
[CDPzを利用したz/OS-ELK連携 - (9)カスタマイズ-fluentdの利用](https://qiita.com/tomotagwork/private/241369db2958fdb0fcf4)

# 利用コンポーネント
今回扱うのは、以下の赤枠の範囲のイメージです。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/132331/a0bf5076-1665-f1a8-2dd6-8fc87cb35801.png)



# 構成

## CDPz側構成

参考: [interactive video: How to create a policy in the Configuration Tool](https://common-data-provider.github.io/create_policy/)

### SYSLOG取得用ポリシーの作成

Create a new policy
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/132331/85fa7afd-225f-7b73-2691-a787c5b11512.png)

名前を指定して DATA STREAMをクリック
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/132331/d6ec804b-e302-74d8-4783-29ead4526c6d.png)

z/OS SYSLOGを選択してSELECT
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/132331/8eff0840-54fd-0d1c-743e-2f03e48dac84.png)

SYSLOG用のDATA STREAMが追加されたので、そのアイコンの中のSubscripbeボタンをクリック
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/132331/865b8868-e280-0543-5e35-88c6e6a7b451.png)

ADD SUBSCRIBERボタンをクリック
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/132331/eedf2a8f-5eab-b58b-edb4-f9fc6be0e2a5.png)

各種値を指定してOK
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/132331/013ce049-ba44-2caa-a0a9-e982f79d03e6.png)

※Logstashの構成では、8081ポートをListenする構成にしていますので、ここではLogstashのホスト名、ポート番号を指定すべきです。しかし今回使用しているテスト環境の制約のため、ホスト=>Linuxへのコネクションが直接張れません。そこで、ポートフォワードの設定をして、ローカルの8081ポートをLinuxの8081へフォワードする設定を別途実施しています。そのため、ここではlocalhost:8081を宛先として指定しています。

UPDATE SUBSCRIPTIONSをクリック
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/132331/4de349fb-53a3-142f-99ff-01c58a1c9b4b.png)

LOG FORWARDERをクリック
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/132331/9bf6d5fb-c78c-8778-7a5e-66bd549bac9e.png)

Log Forwarderがデータを送信する先のData Streamerのポート番号などを指定してOK
参考: [https://www.ibm.com/support/knowledgecenter/ja/SSGE3R_2.1.0/webui_ref_lf.html](https://www.ibm.com/support/knowledgecenter/ja/SSGE3R_2.1.0/webui_ref_lf.html)
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/132331/c49c7006-96a8-2f2b-efbf-b66b8784d5c3.png)

※ZLF_WORKとして、/usr/lpp/IBM/zcdp/v2r1m0/zlf_workを指定したので、このディレクトリを作成しておく。

SAVEをクリック
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/132331/4998ceba-cbfb-1d8b-3a6a-af4fbe6efda8.png)

Policyが作成された
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/132331/d860d228-c3bd-37ab-f133-a925e4eb7d3a.png)

参考: [Output from the Configuration Tool](https://www.ibm.com/support/knowledgecenter/ja/SSGE3R_2.1.0/webui_output.html)

結果、/u/CDP201/cdpConfig/以下に、上の設定に基づいて各種構成ファイルが生成される。

```
[CICS004@EPLEX1:/u/CDP201/cdpConfig] ls -la | grep Test01
-rw-r-----   1 0        IZUADMIN     236 Jan  7 15:53 Test01_syslog.config.properties
-rw-r-----   1 0        IZUADMIN     386 Jan  7 15:53 Test01_syslog.layout
-rw-r-----   1 0        IZUADMIN    3048 Jan  7 15:53 Test01_syslog.policy
-rw-r-----   1 0        IZUADMIN     362 Jan  7 15:53 Test01_syslog.zlf.conf
```

### Data Streamer用構成
Data Streamerのプロシージャー SYS0.PROCLIB(HBODSPRO)を編集して、ポリシー作成時に生成されたTest01_syslog.policyのパスを指定します。必要に応じてData StreamerがListenするポート番号も変更します。

```xxx:SYS0.PROCLIB(HBODSPRO)
...
//STDPARM  DD *                             
PGM /bin/sh                                 
/usr/lpp/IBM/zcdp/v2r1m0/DS/LIB/startup.sh  
/u/CDP201/cdpConfig/Test01_syslog.policy    
50401 start=w trace=n                            
...              
```

### Log Forwarder用構成
参考: [Copying the Log Forwarder configuration files to the ENVDIR directory](https://www.ibm.com/support/knowledgecenter/ja/SSGE3R_2.1.0/lf_config_files.html)

/u/CDP201/cdpConfig以下にTAGというディレクトリを作成し、以下のファイルを名前を変えてコピーします。
/u/CDP201/cdpConfig/Test01_syslog.zlf.conf => /u/CDP201/TAG/EPLEX.ZOS1.zlf.conf
/u/CDP201/cdpConfig/Test01_syslog.config.properties => /u/CDP201/TAG/EPLEX.ZOS1.config.properties
※ファイルの名前はそれぞれ\<sysplex名\>.\<system名\>.xxx とする必要があります。

Log Forwarderのプロシージャー SYS0.PROCLIB(HBOPROC)を編集して、上のファイルを配置したディレクトリをENVDIRに指定します。

```xxx:SYS0.PROCLIB(HBOPROC)
//HBOPROC  PROC OPT=''
// SET HBOBASE='/usr/lpp/IBM/zcdp/v2r1m0/LF/samples'
// SET ENVDIR='-e /u/CDP201/cdpConfig/TAG'
//*
//***** HBOPROC PROC *************************************************
//*
//ZLOGOUT EXEC PGM=BPXBATSL,REGION=0M,TIME=NOLIMIT,
// PARM=('PGM /bin/sh &HBOBASE./startup.sh &OPT &ENVDIR')
//* Output data sets
//STDOUT  DD   SYSOUT=*
//STDERR  DD   SYSOUT=*
//*
//      PEND
```


## ELK側(Linux側)構成

ElasticsearchとKibanaは前回通りの構成で起動しておきます。

### Logstash

/etc/logstash/pipelines.ymlを以下のように編集

```yml:/etc/logstash/pipelines.yml
- pipeline.id: main
  path.config: "/etc/logstash/conf.d_syslog"
```

/etc/logstash/conf.d_syslogディレクトリを作成し、CDPzで提供されているファイルのうち、共通のものとSYSLOG用のフィルターのファイルをこのディレクトリにコピーします。

```
[root@test08 /etc/logstash/conf.d_syslog]# ls -la
合計 24
drwxr-xr-x. 2 root root  142  1月  7 16:05 .
drwxr-xr-x. 5 root root 4096  1月  7 16:03 ..
-rwxr--r--. 1 root root  326  1月  7 16:04 B_CDPz_Input.lsh
-rwxr--r--. 1 root root  469  1月  7 16:04 E_CDPz_Index.lsh
-rw-r--r--. 1 root root  508  1月  7 16:05 H_SYSLOG_or_OPERLOG.lsh
-rw-r--r--. 1 root root  531  1月  7 16:05 N_SYSLOG_or_OPERLOG.lsh
-rwxr--r--. 1 root root  429  1月  7 16:04 Q_CDPz_Elastic.lsh
```

中身確認

```ruby:B_CDPz_Input.lsh

input {
        tcp {
                port => 8081
        }
}

filter {
                json {
                        source => "message"
        }
}
```

```ruby:E_CDPz_Index.lsh
filter {
        split {
        }
        mutate {
                add_field => {
                        "[@metadata][indexname]" => "%{[sourceType]}-%{[sysplexName]}"
                }
        }

        mutate {
                lowercase => [ "[@metadata][indexname]" ]
        }
}
```

```ruby:H_SYSLOG_or_OPERLOG.lsh
filter {
   if [sourceType] == "zOS-SYSLOG-Console" {

      csv{ columns => [  "rcd", "ASID", "TIMESTAMP", "SMFID", "JOBNUM", "CONSOLE", "ROUTECODE", "DESCRIPTOR", "JOBNAME", "FLAGS", "TEXT" ]
         separator => "," }
   }
}
```

```ruby:N_SYSLOG_or_OPERLOG.lsh
filter {
   if [sourceType] == "zOS-SYSLOG-Console" {
      mutate{ add_field => {
         "[@metadata][timestamp]" => "%{TIMESTAMP}"
        }}

      date{ match => [
             "[@metadata][timestamp]", "yyD HH.mm.ss.SSS Z"
        ]}
   }
}
```

```ruby:Q_CDPz_Elastic.lsh
output {
        elasticsearch {
                hosts => [ "localhost" ]
                action => "index"
                index => "cdp-%{[@metadata][indexname]}-%{+yyyyMMdd}"
        }
}
```


# 実行

## ELK側(Linux側)
### Logstash起動
※Elasticsearch, Kibanaは事前に稼働中の想定

とりあえずテストなので手動でlogstashコマンド打って起動させます。

```
# /usr/share/logstash/bin/logstash --path.settings /etc/logstash
OpenJDK 64-Bit Server VM warning: Option UseConcMarkSweepGC was deprecated in version 9.0 and will likely be removed in a future release.
WARNING: An illegal reflective access operation has occurred
WARNING: Illegal reflective access by com.headius.backport9.modules.Modules (file:/usr/share/logstash/logstash-core/lib/jars/jruby-complete-9.2.7.0.jar) to field java.io.FileDescriptor.fd
WARNING: Please consider reporting this to the maintainers of com.headius.backport9.modules.Modules
WARNING: Use --illegal-access=warn to enable warnings of further illegal reflective access operations
WARNING: All illegal access operations will be denied in a future release
Thread.exclusive is deprecated, use Thread::Mutex
Sending Logstash logs to /var/log/logstash which is now configured via log4j2.properties
[2020-01-07T16:47:20,902][INFO ][logstash.runner          ] Starting Logstash {"logstash.version"=>"7.2.0"}
[2020-01-07T16:47:24,030][INFO ][logstash.outputs.elasticsearch] Elasticsearch pool URLs updated {:changes=>{:removed=>[], :added=>[http://localhost:9200/]}}
[2020-01-07T16:47:24,236][WARN ][logstash.outputs.elasticsearch] Restored connection to ES instance {:url=>"http://localhost:9200/"}
[2020-01-07T16:47:24,428][INFO ][logstash.outputs.elasticsearch] ES Output version determined {:es_version=>7}
[2020-01-07T16:47:24,432][WARN ][logstash.outputs.elasticsearch] Detected a 6.x and above cluster: the `type` event field won't be used to determine the document _type {:es_version=>7}
[2020-01-07T16:47:24,468][INFO ][logstash.outputs.elasticsearch] New Elasticsearch output {:class=>"LogStash::Outputs::ElasticSearch", :hosts=>["//localhost"]}
[2020-01-07T16:47:24,572][INFO ][logstash.outputs.elasticsearch] Using default mapping template
[2020-01-07T16:47:24,619][WARN ][org.logstash.instrument.metrics.gauge.LazyDelegatingGauge] A gauge metric of an unknown type (org.jruby.specialized.RubyArrayOneObject) has been create for key: cluster_uuids. This may result in invalid serialization.  It is recommended to log an issue to the responsible developer/development team.
[2020-01-07T16:47:25,127][INFO ][logstash.javapipeline    ] Starting pipeline {:pipeline_id=>"main", "pipeline.workers"=>6, "pipeline.batch.size"=>125, "pipeline.batch.delay"=>50, "pipeline.max_inflight"=>750, :thread=>"#<Thread:0x7df2d573 run>"}
[2020-01-07T16:47:25,127][INFO ][logstash.outputs.elasticsearch] Attempting to install template {:manage_template=>{"index_patterns"=>"logstash-*", "version"=>60001, "settings"=>{"index.refresh_interval"=>"5s", "number_of_shards"=>1}, "mappings"=>{"dynamic_templates"=>[{"message_field"=>{"path_match"=>"message", "match_mapping_type"=>"string", "mapping"=>{"type"=>"text", "norms"=>false}}}, {"string_fields"=>{"match"=>"*", "match_mapping_type"=>"string", "mapping"=>{"type"=>"text", "norms"=>false, "fields"=>{"keyword"=>{"type"=>"keyword", "ignore_above"=>256}}}}}], "properties"=>{"@timestamp"=>{"type"=>"date"}, "@version"=>{"type"=>"keyword"}, "geoip"=>{"dynamic"=>true, "properties"=>{"ip"=>{"type"=>"ip"}, "location"=>{"type"=>"geo_point"}, "latitude"=>{"type"=>"half_float"}, "longitude"=>{"type"=>"half_float"}}}}}}}
[2020-01-07T16:47:25,204][INFO ][logstash.outputs.elasticsearch] Installing elasticsearch template to _template/logstash
[2020-01-07T16:47:25,625][INFO ][logstash.javapipeline    ] Pipeline started {"pipeline.id"=>"main"}
[2020-01-07T16:47:25,645][INFO ][logstash.inputs.tcp      ] Starting tcp input listener {:address=>"0.0.0.0:8081", :ssl_enable=>"false"}
[2020-01-07T16:47:25,742][INFO ][logstash.agent           ] Pipelines running {:count=>1, :running_pipelines=>[:main], :non_running_pipelines=>[]}
[2020-01-07T16:47:26,001][INFO ][logstash.agent           ] Successfully started Logstash API endpoint {:port=>9600}
```

8081ポートがListenされているか確認。

```
[root@test08 /etc/logstash]# netstat -an | grep 8081
tcp6       0      0 :::8081                 :::*                    LISTEN
```


### ポートフォワード設定
※ちょっと特殊な環境なので、Linuxからz/OS USSにssh接続してポートフォワードの設定をします。

```
[root@test08 /etc/logstash]# ssh -f -N -R 8081:localhost:8081 CICS004@eplex1
```

z/OS USSに接続して、8081ポートがListenされているか確認。

```
[CICS004@EPLEX1:/] netstat -a | grep 8081
SSHD7    00787559 127.0.0.1..8081        0.0.0.0..0             Listen
```

## CDPz側

### Data Streamer起動

参考: [Running the Data Streamer](https://www.ibm.com/support/knowledgecenter/ja/SSGE3R_2.1.0/ds_start.html)
Data Stremerを起動します

```                                                            
S HBODSPRO                                                              
$HASP100 HBODSPRO ON STCINRDR                                           
IEF695I START HBODSPRO WITH JOBNAME HBODSPRO IS ASSIGNED TO USER CDP201 
 , GROUP SYS1                                                           
$HASP373 HBODSPRO STARTED                                               
IEF403I HBODSPRO - STARTED - TIME=12.49.39                              
HBO6000I Data Streamer - Version: V2R1M0, Build ID: 1,021, APAR: OA58659
                                                                        
HBO6119I Default Heap:4G Maximum Heap:4G                                
HBO6001I The Data Streamer started successfully in Warm start mode.        

F HBODSPRO,APPL=DISPLAY,POLICY                                           
HBO6076I The current policy is /u/CDP201/cdpConfig/Test01_syslog.policy. 
```

### Log Fowarder起動
参考: [Running the Log Forwarder](https://www.ibm.com/support/knowledgecenter/ja/SSGE3R_2.1.0/lf_start.html)
Log Forwarderを起動します

```
S HBOPROC                                                                 
IEF403I HBOPROC - STARTED - TIME=12.49.51                               
HBOA000I The z/OS Log Forwarder started successfully. 576               
Version: V2R1M0                                                         
Build ID: 20191111-1445                                                 
APAR: OA58659                                                           
HBOC074I The z/OS Log Forwarder uses the local host name EPLEX1 to      
identify this z/OS system.                                              
HBOA002I The z/OS Log Forwarder initialization is complete.             
HBO6003I Gatherer from IP address /127.0.0.1 on port 1026 connected to  
the Data Streamer.                                                         
HBO6059I S-my_elasticsearch-localhost-8081 connected to subscriber      
my_elasticsearch at address localhost/127.0.0.1:8081                                 

F HBOPROC,APPL=DISPLAY,GATHERER,LIST                                  
HBOA032I The following list shows the known data gatherers in the 298 
format 'SIGNATURE STATUS' :                                           
ZOS_SYS_EXIT:SYSLOG STARTED                                           
```

これで、SYSLOGデータがHBOPROCにより拾われて、Data Streamer => LogStash => Elasticsearchという流れで格納されていきます。

# 確認

Kibanaの画面から、Elasticsearchに取り込まれたSYSLOGを確認してみます。

## Elasticsearch Index確認

Kibanaの左側メニューの一番下のManagementを選択 - ElasticsearchのIndex Managementを選択します。
ここで、検索窓にcdp-zosというキーワードを入れてみます。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/132331/779d067d-0487-a836-0d6b-a2da2ce2d475.png)

そうすると、上のようなIndexが作成されていることが確認できます。
(データがElasticsearchに取り込まれて、この名前でIndexが作成されたことが分かります。)

## Intex Pattern作成

次に、KibanaのIndex Patternsを選択し、Create Index Patternsボタンをクリックします。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/132331/09314cfd-38d4-0331-f898-0cc73c2a13d1.png)

Index Patternとして、"cdp-zos-syslog-console-*"を指定し、Next stepボタンをクリックします。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/132331/dd13fea7-7891-1826-1d83-aa35b48274cf.png)

Time filter field nameに、"@timestamp"を指定し、Create index patternボタンをクリックします。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/132331/459ddc1b-6b4f-ca15-af43-78c1d9b93317.png)

Index Patternが作成されました。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/132331/1af53bcf-49bd-fe62-b59b-b7efe6e73075.png)

## Discover画面での確認

左側メニューの一番上のDiscoverを選択し、適宜画面をカスタマイズし、SYSLOGが表示されることを確認します。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/132331/76d26e80-9a82-52bd-d09e-2af515bd769c.png)

これで、一通り、SYSLOGをElasticsearchに取り込んでKibanaで参照する、という所まで確認できました。

# 補足:Data Streamaerから送信されてくるデータの確認

問題判別の参考に、Data StreamerからLogstashに対してどのようなフォーマットでデータが送られてくるのかを確認してみます。

まず、以下のようなLogstashの構成ファイルを作成します。以下は、受け取ったデータをstdoutに出力するようにしています。

```ruby
input{
        tcp{
                port => 8081
        }
}

output {
        stdout { }
}
```

この構成ファイルを使って起動したLogstashに対して、Data StreamerからSYSLOGデータを送信します。

```
{
    "@timestamp" => 2020-01-22T06:02:41.556Z,
          "port" => 59000,
       "message" => "{\"host\":\"EPLEX1\",\"path\":\"SYSLOG\",\"sourceType\":\"zOS-SYSLOG-Console\",\"sourceName\":\"ZOS1-SYSLOG\",\"timeZone\":\"+0900\",\"systemName\":\"ZOS1\",\"sysplexName\":\"EPLEX\",\"inputsequence\":\"20200122060248560:000000\",\"message\":[\"NE,007C,20022 15.02.43.770 +0900,ZOS1    ,STC00809,        ,00000000000000000000000000000000,00000290,CICS0041,00,\\\" IEA630I  OPERATOR CICS004$ NOW ACTIVE,   SYSTEM=ZOS1    , LU=0944FEF3\\\"\\n\",\"NE,007C,20022 15.02.43.770 +0900,ZOS1    ,STC00809,CICS004$,00000000000000000000000000000000,00000290,CICS0041,80,\\\" D T\\\"\\n\",\"NE,007C,20022 15.02.43.770 +0900,ZOS1    ,STC00809,CICS004$,00000000000000000000000000000000,00000090,CICS0041,40,\\\" IEE136I LOCAL: TIME=15.02.43 DATE=2020.022  UTC: TIME=06.02.43 DATE=2020.022\\\"\\n\",\"NE,007C,20022 15.02.44.790 +0900,ZOS1    ,STC00809,        ,00000000000000000000000000000000,00000290,CICS0041,00,\\\" IEA631I  OPERATOR CICS004$ NOW INACTIVE, SYSTEM=ZOS1    , LU=0944FEF3\\\"\\n\"],\"seq\":{\"w\":\"1\",\"c\":\"2\"}}",
      "@version" => "1",
          "host" => "localhost"
}
```

上のJSONデータ中の、"message"フィールドの値が、Data Streamerから送られてきたデータです。JSON形式のデータが送られてきているのが分かります。

次に、Logstash構成で、JSONデータを解釈して受け取るようにしてみます。

```ruby
input {
        tcp {
                port => 8081
        }
}

filter {
                json {
                        source => "message"
        }
}
output {
        stdout { }
}
```

この設定でlogstashを起動しなおすと、Data Streamerから送られてきたデータを標準出力に吐くようになります。これで出力されるデータを見てみると、こんな感じになります。

```
{
      "sysplexName" => "EPLEX",
       "systemName" => "ZOS1",
         "@version" => "1",
          "message" => [
        [0] "NE,0091,20022 15.05.05.820 +0900,ZOS1    ,STC01159,        ,40000000000000000000000000000000,00000090,HBOPROC ,00,\" HBOA002I The z/OS Log Forwarder initialization is complete.\u0000\"\n",
        [1] "NE,006A,20022 15.05.05.840 +0900,ZOS1    ,STC01158,        ,FFFFFFF800FFFFFFFFFFFFFFFFFFFFFF,00000090,HBODSPRO,00,\" HBO6003I Gatherer from IP address /127.0.0.1 on port 1352 connected to the Data Streamer.\u0000\"\n",
        [2] "NE,006A,20022 15.05.06.120 +0900,ZOS1    ,STC01158,        ,FFFFFFF800FFFFFFFFFFFFFFFFFFFFFF,00000090,HBODSPRO,00,\" HBO6059I S-my_elasticsearch-localhost-8081 connected to subscriber my_elasticsearch at address localhost/127.0.0.1:8081 \u0000\"\n",
        [3] "NE,0054,20022 15.05.06.790 +0900,ZOS1    ,STC01065,        ,00000000000000000000000000000000,00000290,CICS0045,00,\" IEA631I  OPERATOR CICS004$ NOW INACTIVE, SYSTEM=ZOS1    , LU=0944FEF3\"\n",
        [4] "NE,0054,20022 15.05.09.480 +0900,ZOS1    ,STC01065,        ,00000000000000000000000000000000,00000290,CICS0046,00,\" IEA630I  OPERATOR CICS004$ NOW ACTIVE,   SYSTEM=ZOS1    , LU=0944FEF3\"\n",
        [5] "NE,0054,20022 15.05.09.480 +0900,ZOS1    ,STC01065,CICS004$,00000000000000000000000000000000,00000290,CICS0046,80,\" D T\"\n",
        [6] "NE,0054,20022 15.05.09.480 +0900,ZOS1    ,STC01065,CICS004$,00000000000000000000000000000000,00000090,CICS0046,40,\" IEE136I LOCAL: TIME=15.05.09 DATE=2020.022  UTC: TIME=06.05.09 DATE=2020.022\"\n",
        [7] "NE,0054,20022 15.05.10.490 +0900,ZOS1    ,STC01065,        ,00000000000000000000000000000000,00000290,CICS0046,00,\" IEA631I  OPERATOR CICS004$ NOW INACTIVE, SYSTEM=ZOS1    , LU=0944FEF3\"\n"
    ],
             "port" => 59002,
       "sourceType" => "zOS-SYSLOG-Console",
             "host" => "EPLEX1",
              "seq" => {
        "w" => "1",
        "c" => "1"
    },
       "sourceName" => "ZOS1-SYSLOG",
         "timeZone" => "+0900",
       "@timestamp" => 2020-01-22T06:05:08.845Z,
             "path" => "SYSLOG",
    "inputsequence" => "20200122060515829:000000"
}
```

このように、Data Streamerからは、1つのJSON構造の中の"message"という要素にいくつかのSYSLOGメッセージがまとめられて送信されてくるようです。
"message"要素は配列になっていて、そこに複数メッセージがCSV形式で格納されているので、それをLogstashのFilterプラグインでフィールド単位に分割してハンドリングすることになります。


# memo

## Log Forwarderのエラー対応

最初、Log Forwarderを起動したら、SYSLOG送信時に以下のようなエラーが発生しました。

```
Jan 07, 2020 5:02:55 PM com.ibm.tivoli.unity.systemz.gatherer.console.ZosConsoleLogGathererImpl gather                              
SEVERE: HBOG050E Because an error occurred with error code 5, z/OS SYSLOG messages are not forwarded to the data streamer for file path SYSLOG. 
```

https://www.ibm.com/support/knowledgecenter/ja/SSGE3R_2.1.0/msg_hbog050e.html

>
Explanation
Some data might be lost due to this error.
The z/OS Log Forwarder continues to gather log records from the most recent data in the data space.
Possible error codes are 5 and 8.
A return code of 5 means that the buffer pointer was not in sync with the data.
A return code of 8 means that some data in the record was corrupted.
>
>Action
If the problem persists, contact IBM Software Support.

なんじゃこれ。
※製品イメージ入手時に一緒に提供されていたPTFのAPPLYし忘れていたので、PTF APPLY実施後に改めて試してみたけどやっぱり同じエラーでNGでした。(PTF適用後は、Build IDの表示が変わってる。)

```xxx:LogForwarderのJOBLOG抜粋
 Jan 09, 2020 3:23:18 PM com.ibm.tivoli.unity.systemz.SystemzLogProvider main
 INFO: HBOA000I The z/OS Log Forwarder started successfully.
 Version: V2R1M0
 Build ID: 20191111-1445
 APAR: OA58659
 Jan 09, 2020 3:23:19 PM com.ibm.tivoli.unity.systemz.common.config.PropertiesConfigurationManagerImpl readConfiguration
 INFO: HBOC001I The z/OS Log Forwarder data configuration file with path name /u/CDP201/cdpConfig/EPLEX.ZOS1.config.properties is loaded and validated.
 Jan 09, 2020 3:23:19 PM com.ibm.tivoli.unity.systemz.common.config.PropertiesConfigurationManagerImpl readConfiguration
 INFO: HBOC074I The z/OS Log Forwarder uses the local host name EPLEX1 to identify this z/OS system.
 Jan 09, 2020 3:23:19 PM com.ibm.tivoli.unity.systemz.controller.LogForwarderController start
 INFO: HBOA002I The z/OS Log Forwarder initialization is complete.
 Jan 09, 2020 3:23:19 PM com.ibm.tivoli.unity.systemz.gatherer.console.ZosConsoleLogGathererImpl gather
 SEVERE: HBOG050E Because an error occurred with error code 5, z/OS SYSLOG messages are not forwarded to the data streamer for file path SYSLOG.
 Jan 09, 2020 3:23:29 PM com.ibm.tivoli.unity.systemz.gatherer.console.ZosConsoleLogGathererImpl gather
 SEVERE: HBOG050E Because an error occurred with error code 5, z/OS SYSLOG messages are not forwarded to the data streamer for file path SYSLOG.
 Jan 09, 2020 3:24:29 PM com.ibm.tivoli.unity.systemz.gatherer.console.ZosConsoleLogGathererImpl gather
 SEVERE: HBOG050E Because an error occurred with error code 5, z/OS SYSLOG messages are not forwarded to the data streamer for file path SYSLOG.
 Jan 09, 2020 3:24:47 PM com.ibm.tivoli.unity.systemz.SystemzLogProvider main
 INFO: HBOA004I The z/OS Log Forwarder stopped successfully.
```

その後、z/OSをIPLして試したらエラーは解消されました。LPA,EXIT辺りが古いバージョンのものと混在していたのかも。