/* Copyright 2017 Telstra Open Source
 *
 *   Licensed 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.
 */

package org.openkilda.wfm.topology.opentsdb;

import org.apache.storm.kafka.spout.KafkaSpout;
import org.apache.storm.tuple.Fields;
import org.openkilda.wfm.topology.opentsdb.bolts.DatapointParseBolt;
import org.slf4j.LoggerFactory;
import org.slf4j.Logger;
import org.apache.storm.generated.StormTopology;
import org.apache.storm.opentsdb.bolt.OpenTsdbBolt;
import org.apache.storm.opentsdb.bolt.TupleOpenTsdbDatapointMapper;
import org.apache.storm.opentsdb.client.OpenTsdbClient;
import org.apache.storm.topology.TopologyBuilder;
import org.openkilda.wfm.ConfigurationException;
import org.openkilda.wfm.LaunchEnvironment;
import org.openkilda.wfm.topology.AbstractTopology;
import org.openkilda.wfm.topology.opentsdb.bolts.OpenTSDBFilterBolt;

import java.util.Collections;

/**
 * Apache Storm topology for sending metrics into Open TSDB.
 */
public class OpenTSDBTopology extends AbstractTopology {

    private static final Logger LOGGER = LoggerFactory.getLogger(OpenTSDBTopology.class);

    public OpenTSDBTopology(LaunchEnvironment env) throws ConfigurationException {
        super(env);
    }

    private final String topic = config.getKafkaOtsdbTopic();
    private final String spoutId = topic + "-spout";
    private final String boltId = topic + "-bolt";
    private final String parseBoltId = topic + "parse-bolt";

    @Override
    public StormTopology createTopology() {
        LOGGER.info("Creating OpenTSDB topology");
        TopologyBuilder tb = new TopologyBuilder();

        checkAndCreateTopic(topic);

        KafkaSpout kafkaSpout = createKafkaSpout(topic, spoutId);
        tb.setSpout(spoutId, kafkaSpout, config.getOpenTsdbNumSpouts());

        tb.setBolt(parseBoltId, new DatapointParseBolt(), config.getGetDatapointParseBoltExecutors())
                .setNumTasks(config.getGetDatapointParseBoltWorkers())
                .shuffleGrouping(spoutId);

        tb.setBolt(boltId, new OpenTSDBFilterBolt(), config.getOpenTsdbFilterBoltExecutors())
                .fieldsGrouping(parseBoltId, new Fields("hash"));

        OpenTsdbClient.Builder tsdbBuilder = OpenTsdbClient
                .newBuilder(config.getOpenTsDBHosts())
//                .sync(config.getOpenTsdbTimeout())
                .sync(30000)
                .returnDetails();
        if(config.isOpenTsdbClientChunkedRequestsEnabled()) {
            tsdbBuilder.enableChunkedEncoding();
        }

        OpenTsdbBolt openTsdbBolt = new OpenTsdbBolt(tsdbBuilder,
                Collections.singletonList(TupleOpenTsdbDatapointMapper.DEFAULT_MAPPER));
        openTsdbBolt.withBatchSize(config.getOpenTsdbBatchSize()).withFlushInterval(config.getOpenTsdbFlushInterval());
//                .failTupleForFailedMetrics();
        tb.setBolt("opentsdb", openTsdbBolt, config.getOpenTsdbBoltExecutors())
                .setNumTasks(config.getOpenTsdbBoltWorkers())
                .shuffleGrouping(boltId);

        return tb.createTopology();
    }

    public static void main(String[] args) {
        try {
            LaunchEnvironment env = new LaunchEnvironment(args);
            (new OpenTSDBTopology(env)).setup();
        } catch (Exception e) {
            System.exit(handleLaunchException(e));
        }
    }
}

@@ -2660,24 +2660,26 @@
 (%22hash%22));%0A%0A
+//
         Open
@@ -2726,16 +2726,18 @@
 bClient%0A
+//
         
@@ -2783,16 +2783,18 @@
 osts())%0A
+//
 //      
@@ -2834,24 +2834,26 @@
 bTimeout())%0A
+//
             
@@ -2869,16 +2869,18 @@
 (30000)%0A
+//
         
@@ -2901,24 +2901,26 @@
 nDetails();%0A
+//
         if(c
@@ -2965,24 +2965,26 @@
 nabled()) %7B%0A
+//
             
@@ -3016,24 +3016,26 @@
 Encoding();%0A
+//
         %7D%0A%0A 
@@ -3028,25 +3028,29 @@
 //        %7D%0A
-%0A
+//%0A//
         Open
@@ -3103,16 +3103,18 @@
 uilder,%0A
+//
         
@@ -3190,24 +3190,26 @@
 T_MAPPER));%0A
+//
         open
@@ -3312,24 +3312,26 @@
 nterval());%0A
+//
 //          
@@ -3362,24 +3362,26 @@
 dMetrics();%0A
+//
         tb.s
@@ -3440,32 +3440,34 @@
 oltExecutors())%0A
+//
                 
@@ -3504,32 +3504,34 @@
 bBoltWorkers())%0A
+//
                 

remove opentsdb writer bolt