| # Licensed to the Apache Software Foundation (ASF) under one |
| # or more contributor license agreements. See the NOTICE file |
| # distributed with this work for additional information |
| # regarding copyright ownership. The ASF licenses this file |
| # to you under the Apache License, Version 2.0 (the |
| # "License"); you may not use this file except in compliance |
| # with the License. You may obtain a copy of the License at |
| # |
| # http://www.apache.org/licenses/LICENSE-2.0 |
| # |
| # Unless required by applicable law or agreed to in writing, software |
| # distributed under the License is distributed on an "AS IS" BASIS, |
| # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| # See the License for the specific language governing permissions and |
| # limitations under the License. |
| |
| name: "yaf" |
| config: |
| topology.workers: 1 |
| |
| components: |
| - id: "yafParser" |
| className: "org.apache.metron.parsing.parsers.BasicYafParser" |
| - id: "genericMessageFilter" |
| className: "org.apache.metron.filters.GenericMessageFilter" |
| - id: "indexAdapter" |
| className: "org.apache.metron.indexing.adapters.ESTimedRotatingAdapter" |
| - id: "metricConfig" |
| className: "org.apache.commons.configuration.BaseConfiguration" |
| configMethods: |
| - name: "setProperty" |
| args: |
| - "org.apache.metron.metrics.reporter.graphite" |
| - "${org.apache.metron.metrics.reporter.graphite}" |
| - name: "setProperty" |
| args: |
| - "org.apache.metron.metrics.reporter.console" |
| - "${org.apache.metron.metrics.reporter.console}" |
| - name: "setProperty" |
| args: |
| - "org.apache.metron.metrics.reporter.jmx" |
| - "${org.apache.metron.metrics.reporter.jmx}" |
| - name: "setProperty" |
| args: |
| - "org.apache.metron.metrics.graphite.address" |
| - "${org.apache.metron.metrics.graphite.address}" |
| - name: "setProperty" |
| args: |
| - "org.apache.metron.metrics.graphite.port" |
| - "${org.apache.metron.metrics.graphite.port}" |
| - name: "setProperty" |
| args: |
| - "org.apache.metron.metrics.TelemetryParserBolt.acks" |
| - "${org.apache.metron.metrics.TelemetryParserBolt.acks}" |
| - name: "setProperty" |
| args: |
| - "org.apache.metron.metrics.TelemetryParserBolt.emits" |
| - "${org.apache.metron.metrics.TelemetryParserBolt.emits}" |
| - name: "setProperty" |
| args: |
| - "org.apache.metron.metrics.TelemetryParserBolt.fails" |
| - "${org.apache.metron.metrics.TelemetryParserBolt.fails}" |
| - name: "setProperty" |
| args: |
| - "org.apache.metron.metrics.GenericEnrichmentBolt.acks" |
| - "${org.apache.metron.metrics.GenericEnrichmentBolt.acks}" |
| - name: "setProperty" |
| args: |
| - "org.apache.metron.metrics.GenericEnrichmentBolt.emits" |
| - "${org.apache.metron.metrics.GenericEnrichmentBolt.emits}" |
| - name: "setProperty" |
| args: |
| - "org.apache.metron.metrics.GenericEnrichmentBolt.fails" |
| - "${org.apache.metron.metrics.GenericEnrichmentBolt.fails}" |
| - name: "setProperty" |
| args: |
| - "org.apache.metron.metrics.TelemetryIndexingBolt.acks" |
| - "${org.apache.metron.metrics.TelemetryIndexingBolt.acks}" |
| - name: "setProperty" |
| args: |
| - "org.apache.metron.metrics.TelemetryIndexingBolt.emits" |
| - "${org.apache.metron.metrics.TelemetryIndexingBolt.emits}" |
| - name: "setProperty" |
| args: |
| - "org.apache.metron.metrics.TelemetryIndexingBolt.fails" |
| - "${org.apache.metron.metrics.TelemetryIndexingBolt.fails}" |
| - id: "zkHosts" |
| className: "storm.kafka.ZkHosts" |
| constructorArgs: |
| - "${kafka.zk}" |
| - id: "kafkaConfig" |
| className: "storm.kafka.SpoutConfig" |
| constructorArgs: |
| # zookeeper hosts |
| - ref: "zkHosts" |
| # topic name |
| - "${spout.kafka.topic.yaf}" |
| # zk root |
| - "" |
| # id |
| - "${spout.kafka.topic.yaf}" |
| properties: |
| - name: "forceFromStart" |
| value: true |
| - name: "startOffsetTime" |
| value: -1 |
| |
| spouts: |
| - id: "kafkaSpout" |
| className: "storm.kafka.KafkaSpout" |
| constructorArgs: |
| - ref: "kafkaConfig" |
| |
| bolts: |
| - id: "parserBolt" |
| className: "org.apache.metron.bolt.TelemetryParserBolt" |
| configMethods: |
| - name: "withMessageParser" |
| args: |
| - ref: "yafParser" |
| - id: "indexingBolt" |
| className: "org.apache.metron.indexing.TelemetryIndexingBolt" |
| configMethods: |
| - name: "withIndexIP" |
| args: |
| - "${es.ip}" |
| - name: "withIndexPort" |
| args: |
| - ${es.port} |
| - name: "withClusterName" |
| args: |
| - "${es.clustername}" |
| - name: "withIndexName" |
| args: |
| - "yaf_index" |
| - name: "withIndexTimestamp" |
| args: |
| - "yyyy.MM.dd.hh" |
| - name: "withDocumentName" |
| args: |
| - "yaf_doc" |
| - name: "withBulk" |
| args: |
| - 1 |
| - name: "withIndexAdapter" |
| args: |
| - ref: "indexAdapter" |
| - name: "withMetricConfiguration" |
| args: |
| - ref: "metricConfig" |
| - id: "errorIndexingBolt" |
| className: "org.apache.metron.indexing.TelemetryIndexingBolt" |
| configMethods: |
| - name: "withIndexIP" |
| args: |
| - "${es.ip}" |
| - name: "withIndexPort" |
| args: |
| - ${es.port} |
| - name: "withClusterName" |
| args: |
| - "${es.clustername}" |
| - name: "withIndexName" |
| args: |
| - "error" |
| - name: "withIndexTimestamp" |
| args: |
| - "yyyy.MM" |
| - name: "withDocumentName" |
| args: |
| - "yaf_error" |
| - name: "withBulk" |
| args: |
| - 1 |
| - name: "withIndexAdapter" |
| args: |
| - ref: "indexAdapter" |
| - name: "withMetricConfiguration" |
| args: |
| - ref: "metricConfig" |
| |
| streams: |
| - name: "spout -> parser" |
| from: "kafkaSpout" |
| to: "parserBolt" |
| grouping: |
| type: SHUFFLE |
| - name: "parser -> indexing" |
| from: "parserBolt" |
| to: "indexingBolt" |
| grouping: |
| streamId: "message" |
| type: FIELDS |
| args: ["key"] |
| - name: "parser -> errors" |
| from: "parserBolt" |
| to: "errorIndexingBolt" |
| grouping: |
| streamId: "error" |
| type: SHUFFLE |
| - name: "indexing -> errors" |
| from: "indexingBolt" |
| to: "errorIndexingBolt" |
| grouping: |
| streamId: "error" |
| type: SHUFFLE |