components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/util/DruidSupervisorGenerator.java

   1 /*\r
   2  * ============LICENSE_START=======================================================\r
   3  * ONAP : DataLake\r
   4  * ================================================================================\r
   5  * Copyright 2019 China Mobile\r
   6  *=================================================================================\r
   7  * Licensed under the Apache License, Version 2.0 (the "License");\r
   8  * you may not use this file except in compliance with the License.\r
   9  * You may obtain a copy of the License at\r
  10  *\r
  11  *     http://www.apache.org/licenses/LICENSE-2.0\r
  12  *\r
  13  * Unless required by applicable law or agreed to in writing, software\r
  14  * distributed under the License is distributed on an "AS IS" BASIS,\r
  15  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\r
  16  * See the License for the specific language governing permissions and\r
  17  * limitations under the License.\r
  18  * ============LICENSE_END=========================================================\r
  19  */\r
  20 package org.onap.datalake.feeder.util;\r
  21 \r
  22 import com.fasterxml.jackson.databind.JsonNode;\r
  23 import com.fasterxml.jackson.databind.ObjectMapper;\r
  24 import com.fasterxml.jackson.databind.node.JsonNodeType;\r
  25 import lombok.Getter;\r
  26 import org.apache.velocity.Template;\r
  27 import org.apache.velocity.VelocityContext;\r
  28 import org.apache.velocity.app.Velocity;\r
  29 import org.apache.velocity.runtime.RuntimeConstants;\r
  30 import org.apache.velocity.runtime.resource.loader.ClasspathResourceLoader;\r
  31 import org.slf4j.Logger;\r
  32 import org.slf4j.LoggerFactory;\r
  33 \r
  34 import java.io.BufferedWriter;\r
  35 import java.io.FileWriter;\r
  36 import java.io.IOException;\r
  37 import java.net.MalformedURLException;\r
  38 import java.util.ArrayList;\r
  39 import java.util.Iterator;\r
  40 import java.util.List;\r
  41 import java.util.Map.Entry;\r
  42 \r
  43 \r
  44 /*\r
  45  * read sample json and output supervisor to  resources\druid\generated\r
  46  * need manual edit to be production ready, final versions are in resources\druid\r
  47  *\r
  48  * http://druid.io/docs/latest/tutorials/tutorial-ingestion-spec.html\r
  49  * http://druid.io/docs/latest/ingestion/flatten-json\r
  50  *\r
  51  *\r
  52  * todo:\r
  53  * reduce the manual editing\r
  54  * path hard coded\r
  55  * auto get topics,\r
  56  * auto get sample, and for each topic, get multiple samples.\r
  57  * make supervisor file names consistent\r
  58  * dimension type default is string, in msgrtr.apinode.metrics.dmaap , many are long/double, so need to generate dimensionsSpec, this is done at the end of printFlattenSpec()\r
  59  */\r
  60 \r
  61 @Getter\r
  62 public class DruidSupervisorGenerator {\r
  63 \r
  64     private final Logger log = LoggerFactory.getLogger(this.getClass());\r
  65 \r
  66     Template template = null;\r
  67     VelocityContext context;\r
  68 \r
  69     List<String[]> dimensions;\r
  70 \r
  71     public DruidSupervisorGenerator() {\r
  72         dimensions = new ArrayList<>();\r
  73 \r
  74         Velocity.setProperty(RuntimeConstants.RESOURCE_LOADER, "classpath");\r
  75         Velocity.setProperty("classpath.resource.loader.class", ClasspathResourceLoader.class.getName());\r
  76 \r
  77         Velocity.init();\r
  78 \r
  79         context = new VelocityContext();\r
  80 \r
  81         context.put("host", "message-router-kafka:9092");//TODO get from config\r
  82 \r
  83         template = Velocity.getTemplate("druid/kafka-supervisor-template.vm");\r
  84     }\r
  85 \r
  86     private void printNode(String prefix, JsonNode node) {\r
  87 \r
  88         // lets see what type the node is\r
  89         //              System.out.println("NodeType=" + node.getNodeType() + ", isContainerNode=" + node.isContainerNode() + ", " + node); // prints OBJECT\r
  90 \r
  91         if (node.isContainerNode()) {\r
  92 \r
  93             Iterator<Entry<String, JsonNode>> fields = node.fields();\r
  94 \r
  95             while (fields.hasNext()) {\r
  96                 Entry<String, JsonNode> field = fields.next();\r
  97                 printNode(prefix + "." + field.getKey(), field.getValue());\r
  98             }\r
  99 \r
 100             if (node.isArray()) {\r
 101                 Iterator<JsonNode> elements = node.elements();\r
 102                 int i = 0;\r
 103                 while (elements.hasNext()) {\r
 104                     JsonNode element = elements.next();\r
 105                     printNode(prefix + "[" + i + "]", element);\r
 106                     i++;\r
 107                 }\r
 108             }\r
 109 \r
 110         } else {\r
 111             printFlattenSpec(node.getNodeType(), prefix);\r
 112         }\r
 113 \r
 114     }\r
 115 \r
 116     private void printFlattenSpec(JsonNodeType type, String path) {\r
 117         String name = path.substring(2).replace('.', ':');\r
 118         // lets see what type the node is\r
 119         log.info("{");\r
 120         log.info("\"type\": \"path\",");\r
 121         log.info("\"name\": \"" + name + "\",");\r
 122         log.info("\"expr\": \"" + path + "\"");\r
 123         log.info("},");\r
 124 \r
 125         dimensions.add(new String[]{name, path});\r
 126     }\r
 127 \r
 128     public void doTopic(String topic) throws IOException {\r
 129         dimensions.clear();\r
 130 \r
 131         String sampleFileName = "src/main/resources/druid/" + topic + "-sample-format.json";//FIXME hard coded path\r
 132         String outputFileName = "src/main/resources/druid/generated/" + topic + "-kafka-supervisor.json";\r
 133 \r
 134         // Get the contents of json as a string using commons IO IOUTils class.\r
 135         String sampleJson = Util.getTextFromFile(sampleFileName);\r
 136 \r
 137         // create an ObjectMapper instance.\r
 138         ObjectMapper mapper = new ObjectMapper();\r
 139         // use the ObjectMapper to read the json string and create a tree\r
 140         JsonNode root = mapper.readTree(sampleJson);\r
 141         printNode("$", root);\r
 142 \r
 143         context.put("topic", topic);\r
 144         context.put("timestamp", "event-header:timestamp");//FIXME hard coded, should be topic based\r
 145         context.put("timestampFormat", "yyyyMMdd-HH:mm:ss:SSS");//FIXME hard coded, should be topic based\r
 146         context.put("dimensions", dimensions);\r
 147 \r
 148         BufferedWriter out = new BufferedWriter(new FileWriter(outputFileName));\r
 149 \r
 150         template.merge(context, out);\r
 151         out.close();\r
 152     }\r
 153 \r
 154     public static void main(String[] args) throws MalformedURLException, IOException {\r
 155         String[] topics = new String[]{"AAI-EVENT", "msgrtr.apinode.metrics.dmaap", "unauthenticated.DCAE_CL_OUTPUT", "unauthenticated.SEC_FAULT_OUTPUT"};//FIXME hard coded\r
 156 \r
 157         DruidSupervisorGenerator p = new DruidSupervisorGenerator();\r
 158 \r
 159         for (String topic : topics) {\r
 160             p.doTopic(topic);\r
 161         }\r
 162     }\r
 163 }\r