{"action":"created","issue":{"url":"https://api.github.com/repos/mesos/kafka/issues/175","labels_url":"https://api.github.com/repos/mesos/kafka/issues/175/labels{/name}","comments_url":"https://api.github.com/repos/mesos/kafka/issues/175/comments","events_url":"https://api.github.com/repos/mesos/kafka/issues/175/events","html_url":"https://github.com/mesos/kafka/pull/175","id":127272702,"number":175,"title":"Basic broker metrics","user":{"login":"olegkovalenko","id":350927,"avatar_url":"https://avatars.githubusercontent.com/u/350927?v=3","gravatar_id":"","url":"https://api.github.com/users/olegkovalenko","html_url":"https://github.com/olegkovalenko","followers_url":"https://api.github.com/users/olegkovalenko/followers","following_url":"https://api.github.com/users/olegkovalenko/following{/other_user}","gists_url":"https://api.github.com/users/olegkovalenko/gists{/gist_id}","starred_url":"https://api.github.com/users/olegkovalenko/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/olegkovalenko/subscriptions","organizations_url":"https://api.github.com/users/olegkovalenko/orgs","repos_url":"https://api.github.com/users/olegkovalenko/repos","events_url":"https://api.github.com/users/olegkovalenko/events{/privacy}","received_events_url":"https://api.github.com/users/olegkovalenko/received_events","type":"User","site_admin":false},"labels":[],"state":"open","locked":false,"assignee":null,"milestone":null,"comments":2,"created_at":"2016-01-18T17:16:49Z","updated_at":"2016-01-19T11:54:16Z","closed_at":null,"pull_request":{"url":"https://api.github.com/repos/mesos/kafka/pulls/175","html_url":"https://github.com/mesos/kafka/pull/175","diff_url":"https://github.com/mesos/kafka/pull/175.diff","patch_url":"https://github.com/mesos/kafka/pull/175.patch"},"body":"Motivation:\r\nMetrics are useful and important information for monitoring broker.\r\nImportant metrics: \r\n\r\n> 1 kafka.server:type=ReplicaManager,name=UnderReplicatedPartitions\r\n> number of under-replicated partitions (| ISR | < | all replicas |) (dangerous when 0).\r\n> 2 kafka.controller:type=KafkaController,name=OfflinePartitionsCount\r\n> number of partitions that don't have an active leader and are hence not writable or readable (dangerous when 0).\r\n> 3 kafka.controller:type=KafkaController,name=ActiveControllerCount\r\n> number of active controllers in the cluster. (dangerous when value is anything other than 1).\r\n\r\nfor more information on metrics see http://docs.confluent.io/1.0/kafka/monitoring.html\r\n\r\nChanges:\r\n- executor: executor start to pull metrics every 5 seconds (right after broker start)\r\nthrough JMX and send them to scheduler via sendFrameworkMessage\r\n- scheduler: scheduler receives servialized JSON metrics and updates\r\ncorresponding active broker\r\n- cli: metrics are shown in output of `broker list`, for example:\r\n```\r\n./kafka-mesos.sh broker list\r\nbrokers:\r\n  id: 0\r\n  active: true\r\n  state: running\r\n  ...\r\n  metrics:\r\n    collected: 2016-01-18 11:53:36Z\r\n    under-replicated-partitions: 0\r\n    offline-partitions-count: 0\r\n    active-controller-count: 1\r\n\r\n  id: 1\r\n  active: true\r\n  state: running\r\n  ...\r\n  metrics:\r\n    collected: 2016-01-18 11:53:36Z\r\n    under-replicated-partitions: 0\r\n    offline-partitions-count: 0\r\n    active-controller-count: 0\r\n```\r\n\r\n"},"comment":{"url":"https://api.github.com/repos/mesos/kafka/issues/comments/172828162","html_url":"https://github.com/mesos/kafka/pull/175#issuecomment-172828162","issue_url":"https://api.github.com/repos/mesos/kafka/issues/175","id":172828162,"user":{"login":"joestein","id":115151,"avatar_url":"https://avatars.githubusercontent.com/u/115151?v=3","gravatar_id":"","url":"https://api.github.com/users/joestein","html_url":"https://github.com/joestein","followers_url":"https://api.github.com/users/joestein/followers","following_url":"https://api.github.com/users/joestein/following{/other_user}","gists_url":"https://api.github.com/users/joestein/gists{/gist_id}","starred_url":"https://api.github.com/users/joestein/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/joestein/subscriptions","organizations_url":"https://api.github.com/users/joestein/orgs","repos_url":"https://api.github.com/users/joestein/repos","events_url":"https://api.github.com/users/joestein/events{/privacy}","received_events_url":"https://api.github.com/users/joestein/received_events","type":"User","site_admin":false},"created_at":"2016-01-19T11:54:16Z","updated_at":"2016-01-19T11:54:16Z","body":"We should also consider too another option (which has come up a few times since folks very often end up doing this in some fashion anyways) is to have the metrics and logs get written to topics (optionally to another broker cluster running (for now, with a different scheduler). In any case it should be a different zookeeper ensemble cluster. Brokers emit lots of logs and metrics and having them \"just go\" to a topic defined with a broker list as a feature flag on the CLI (logs == active with topic and broker list to supply (if no broker list then existing cluster) and metrics == active same (but likely different topics).\r\n\r\nif we want to make it best integrated I would suggest too that write the metric and log data with an avro message like log line is pretty standard starting point for that.\r\n\r\n\r\nwe could put into the mesos/kafka code base something like\r\n\r\n```\r\n/**\r\n * Licensed to the Apache Software Foundation (ASF) under one or more\r\n * contributor license agreements.  See the NOTICE file distributed with\r\n * this work for additional information regarding copyright ownership.\r\n * The ASF licenses this file to You under the Apache License, Version 2.0\r\n * (the \"License\"); you may not use this file except in compliance with\r\n * the License.  You may obtain a copy of the License at\r\n *\r\n *    http://www.apache.org/licenses/LICENSE-2.0\r\n *\r\n * Unless required by applicable law or agreed to in writing, software\r\n * distributed under the License is distributed on an \"AS IS\" BASIS,\r\n * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\r\n * See the License for the specific language governing permissions and\r\n * limitations under the License.\r\n */\r\n\r\npackage ly.stealth.kafka.metrics.yammer;\r\n\r\nimport com.yammer.metrics.core.*;\r\nimport com.yammer.metrics.reporting.AbstractPollingReporter;\r\nimport com.yammer.metrics.stats.Snapshot;\r\nimport io.confluent.kafka.serializers.KafkaAvroSerializer;\r\nimport ly.stealth.avro.JsonToAvroConverter;\r\nimport org.apache.avro.generic.IndexedRecord;\r\nimport org.apache.kafka.clients.producer.KafkaProducer;\r\nimport org.apache.kafka.clients.producer.ProducerConfig;\r\nimport org.apache.kafka.clients.producer.ProducerRecord;\r\nimport org.json.JSONException;\r\nimport org.json.JSONObject;\r\n\r\nimport java.util.Map;\r\nimport java.util.Properties;\r\n\r\n/**\r\n * KafkaReporter produces all metrics data from a provided registry to Kafka topic with a given reporting interval\r\n * encoded as Avro.\r\n * Avro schema will be created dynamically to match JSON representation of the registry. LogLine field will also\r\n * be added to contain simple metadata about metrics. LogLines logtypeid will be 7 (metrics data) and the source\r\n * will be \"metrics\".\r\n */\r\npublic class KafkaReporter extends AbstractPollingReporter implements MetricProcessor<JSONObject> {\r\n    private final KafkaProducer<String, IndexedRecord> producer;\r\n    private final MetricPredicate predicate = MetricPredicate.ALL;\r\n    private final JsonToAvroConverter converter;\r\n    private final String topic;\r\n\r\n    public KafkaReporter(MetricsRegistry metricsRegistry, Properties producerProperties, String topic) {\r\n        super(metricsRegistry, \"kafka-topic-reporter\");\r\n\r\n        producerProperties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, KafkaAvroSerializer.class);\r\n        producerProperties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, KafkaAvroSerializer.class);\r\n\r\n        this.producer = new KafkaProducer<String, IndexedRecord>(producerProperties);\r\n        this.converter = new JsonToAvroConverter();\r\n        this.topic = topic;\r\n\r\n    }\r\n\r\n    public void run() {\r\n        try {\r\n            JSONObject object = new JSONObject();\r\n            for (Map.Entry<MetricName, Metric> entry : getMetricsRegistry().allMetrics().entrySet()) {\r\n                if (predicate.matches(entry.getKey(), entry.getValue())) {\r\n                    entry.getValue().processWith(this, entry.getKey(), object);\r\n                }\r\n            }\r\n            ProducerRecord<String, IndexedRecord> record = new ProducerRecord<String, IndexedRecord>(this.topic, converter.convert(object.toString()));\r\n            this.producer.send(record);\r\n        } catch (Exception e) {\r\n            throw new RuntimeException(e.getMessage(), e);\r\n        }\r\n    }\r\n\r\n    @Override\r\n    public void processGauge(MetricName name, Gauge<?> gauge, JSONObject context) {\r\n        JSONObject jsonObject = new JSONObject();\r\n        try {\r\n            jsonObject.put(\"name\", getName(name));\r\n            jsonObject.put(\"type\", \"gauge\");\r\n            jsonObject.put(\"value\", gauge.value());\r\n            context.put(name.getName(), jsonObject);\r\n        } catch (JSONException e) {\r\n            throw new RuntimeException(e);\r\n        }\r\n    }\r\n\r\n    @Override\r\n    public void processCounter(MetricName name, Counter counter, JSONObject context) {\r\n        JSONObject jsonObject = new JSONObject();\r\n        try {\r\n            jsonObject.put(\"name\", getName(name));\r\n            jsonObject.put(\"type\", \"counter\");\r\n            jsonObject.put(\"value\", counter.count());\r\n            context.put(name.getName(), jsonObject);\r\n        } catch (JSONException e) {\r\n            throw new RuntimeException(e);\r\n        }\r\n    }\r\n\r\n    @Override\r\n    public void processMeter(MetricName name, Metered meter, JSONObject context) {\r\n        JSONObject jsonObject = new JSONObject();\r\n        try {\r\n            jsonObject.put(\"name\", getName(name));\r\n            jsonObject.put(\"type\", \"meter\");\r\n            JSONObject meterJsonObject = new JSONObject();\r\n\r\n            addMeterInfo(meter, meterJsonObject);\r\n\r\n            jsonObject.put(\"value\", meterJsonObject);\r\n\r\n            context.put(name.getName(), jsonObject);\r\n        } catch (JSONException e) {\r\n            throw new RuntimeException(e);\r\n        }\r\n    }\r\n\r\n    private void addMeterInfo(Metered meter, JSONObject meterJsonObject) throws JSONException {\r\n        meterJsonObject.put(\"rateUnit\", meter.rateUnit());\r\n        meterJsonObject.put(\"eventType\", meter.eventType());\r\n        meterJsonObject.put(\"count\", meter.count());\r\n        meterJsonObject.put(\"meanRate\", meter.meanRate());\r\n        meterJsonObject.put(\"oneMinuteRate\", meter.oneMinuteRate());\r\n        meterJsonObject.put(\"fiveMinuteRate\", meter.fiveMinuteRate());\r\n        meterJsonObject.put(\"fifteenMinuteRate\", meter.fifteenMinuteRate());\r\n    }\r\n\r\n    @Override\r\n    public void processHistogram(MetricName name, Histogram histogram, JSONObject context) {\r\n        JSONObject jsonObject = new JSONObject();\r\n        try {\r\n            jsonObject.put(\"name\", getName(name));\r\n            jsonObject.put(\"type\", \"meter\");\r\n            JSONObject histogramJsonObject = new JSONObject();\r\n\r\n            histogramJsonObject.put(\"min\", histogram.min());\r\n            histogramJsonObject.put(\"max\", histogram.max());\r\n            histogramJsonObject.put(\"mean\", histogram.mean());\r\n            histogramJsonObject.put(\"stdDev\", histogram.stdDev());\r\n\r\n            Snapshot snapshot = histogram.getSnapshot();\r\n            JSONObject snapshotJsonObject = new JSONObject();\r\n            snapshotJsonObject.put(\"median\", snapshot.getMedian());\r\n            snapshotJsonObject.put(\"75%\", snapshot.get75thPercentile());\r\n            snapshotJsonObject.put(\"95%\", snapshot.get95thPercentile());\r\n            snapshotJsonObject.put(\"98%\", snapshot.get98thPercentile());\r\n            snapshotJsonObject.put(\"99%\", snapshot.get99thPercentile());\r\n            snapshotJsonObject.put(\"99.9%\", snapshot.get999thPercentile());\r\n\r\n            histogramJsonObject.put(\"snapshot\", snapshotJsonObject);\r\n\r\n            jsonObject.put(\"value\", histogramJsonObject);\r\n            context.put(name.getName(), jsonObject);\r\n        } catch (JSONException e) {\r\n            throw new RuntimeException(e);\r\n        }\r\n    }\r\n\r\n    @Override\r\n    public void processTimer(MetricName name, Timer timer, JSONObject context) {\r\n        JSONObject jsonObject = new JSONObject();\r\n        try {\r\n            jsonObject.put(\"name\", getName(name));\r\n            jsonObject.put(\"type\", \"meter\");\r\n            JSONObject timerJsonObject = new JSONObject();\r\n\r\n            timerJsonObject.put(\"unit\", timer.durationUnit());\r\n            timerJsonObject.put(\"min\", timer.min());\r\n            timerJsonObject.put(\"max\", timer.max());\r\n            timerJsonObject.put(\"mean\", timer.mean());\r\n            timerJsonObject.put(\"stdDev\", timer.stdDev());\r\n            addMeterInfo(timer, timerJsonObject);\r\n\r\n            Snapshot snapshot = timer.getSnapshot();\r\n            JSONObject snapshotJsonObject = new JSONObject();\r\n            snapshotJsonObject.put(\"median\", snapshot.getMedian());\r\n            snapshotJsonObject.put(\"75%\", snapshot.get75thPercentile());\r\n            snapshotJsonObject.put(\"95%\", snapshot.get95thPercentile());\r\n            snapshotJsonObject.put(\"98%\", snapshot.get98thPercentile());\r\n            snapshotJsonObject.put(\"99%\", snapshot.get99thPercentile());\r\n            snapshotJsonObject.put(\"99.9%\", snapshot.get999thPercentile());\r\n\r\n            timerJsonObject.put(\"snapshot\", snapshotJsonObject);\r\n\r\n            jsonObject.put(\"value\", timerJsonObject);\r\n\r\n            context.put(name.getName(), jsonObject);\r\n        } catch (JSONException e) {\r\n            throw new RuntimeException(e);\r\n        }\r\n    }\r\n\r\n    private JSONObject getName(MetricName metricName) throws JSONException {\r\n        String group = metricName.getGroup();\r\n        String name = metricName.getName();\r\n        String scope = metricName.getScope();\r\n        String type = metricName.getType();\r\n        JSONObject jsonObject = new JSONObject();\r\n        jsonObject.put(\"name\", name);\r\n        jsonObject.put(\"group\", group);\r\n        jsonObject.put(\"scope\", scope);\r\n        jsonObject.put(\"type\", type);\r\n        return jsonObject;\r\n    }\r\n\r\n    @Override\r\n    public void shutdown() {\r\n        try {\r\n            super.shutdown();\r\n        } finally {\r\n            this.producer.close();\r\n        }\r\n    }\r\n}\r\n```\r\n\r\nand then have the mesos/kafka code run in the broker on startup so we can then can set the properties on the broker for the metrics to be reported to the broker and cluster without having to need knowledge or take time to thread that together every single time you do a setup. I would take that and this\r\n\r\n```\r\npackage ly.stealth.kafka.emitters;\r\n\r\nimport io.confluent.kafka.serializers.KafkaAvroSerializer;\r\nimport ly.stealth.avro.LogLine;\r\nimport org.apache.avro.generic.IndexedRecord;\r\nimport org.apache.kafka.clients.producer.KafkaProducer;\r\nimport org.apache.kafka.clients.producer.ProducerConfig;\r\nimport org.apache.kafka.clients.producer.ProducerRecord;\r\nimport org.apache.kafka.common.config.ConfigException;\r\nimport org.apache.log4j.AppenderSkeleton;\r\nimport org.apache.log4j.Level;\r\nimport org.apache.log4j.spi.LoggingEvent;\r\n\r\nimport java.util.HashMap;\r\nimport java.util.Map;\r\nimport java.util.Properties;\r\n\r\n/**\r\n * KafkaLogAppender produces all log events to Kafka topic encoded as Avro.\r\n * Avro schema is defined as follows:\r\n * {\r\n *     \"type\": \"record\",\r\n *     \"name\": \"ly.stealth.avro.LogLine\",\r\n *     \"fields\": [\r\n *         {\r\n *             \"name\": \"line\",\r\n *             \"type\": [\"null\", \"string\"],\r\n *             \"default\": null\r\n *         },\r\n *         {\r\n *             \"name\": \"source\",\r\n *             \"type\": [\"null\", \"string\"],\r\n *             \"default\": null\r\n *         },\r\n *         {\r\n *             \"name\": \"tag\",\r\n *             \"type\": [\r\n *                 \"null\",\r\n *                 {\r\n *                     \"type\": \"map\",\r\n *                     \"values\": \"string\"\r\n *                 }\r\n *             ],\r\n *             \"default\": null\r\n *         },\r\n *         {\r\n *             \"name\": \"logtypeid\",\r\n *             \"type\": [\"null\", \"long\"],\r\n *             \"default\": null\r\n *         },\r\n *         {\r\n *             \"name\": \"timings\",\r\n *             \"type\": [\r\n *                 \"null\",\r\n *                 {\r\n *                     \"type\": \"map\",\r\n *                     \"values\": \"long\"\r\n *                 }\r\n *             ],\r\n *             \"default\": null\r\n *         }\r\n *     ]\r\n * }\r\n *\r\n * \"line\" field will contain the actual log message.\r\n * \"logtypeid\" will be the log level:\r\n *      1 - Trace\r\n *      2 - Debug\r\n *      3 - Info\r\n *      4 - Warn\r\n *      5 - Error\r\n *      6 - Fatal\r\n * \"source\" will be the thread name that logged the message.\r\n * \"tag\" will contain 2 entries:\r\n *      \"logger.name\" - logger name, e.g. the one that is used in call Logger.getLogger()\r\n *      \"location\" - event location information, e.g. source code line that logged the message\r\n * \"timings\" will contain 1 entry:\r\n *      \"emitted\" - timestamp in milliseconds when the message was sent\r\n */\r\npublic class KafkaLogAppender extends AppenderSkeleton {\r\n    public static final Long TraceLogTypeId = 1L;\r\n    public static final Long DebugLogTypeId = 2L;\r\n    public static final Long InfoLogTypeId = 3L;\r\n    public static final Long WarnLogTypeId = 4L;\r\n    public static final Long ErrorLogTypeId = 5L;\r\n    public static final Long FatalLogTypeId = 6L;\r\n\r\n    private static final Map<Level, Long> logLevels = new HashMap<Level, Long>();\r\n\r\n    static {\r\n        logLevels.put(Level.TRACE, TraceLogTypeId);\r\n        logLevels.put(Level.DEBUG, DebugLogTypeId);\r\n        logLevels.put(Level.INFO, InfoLogTypeId);\r\n        logLevels.put(Level.WARN, WarnLogTypeId);\r\n        logLevels.put(Level.ERROR, ErrorLogTypeId);\r\n        logLevels.put(Level.FATAL, FatalLogTypeId);\r\n    }\r\n\r\n    private KafkaProducer<String, IndexedRecord> producer;\r\n\r\n    private String topic;\r\n    private String brokerList;\r\n    private String schemaRegistryUrl;\r\n\r\n    public KafkaLogAppender() {\r\n\r\n    }\r\n\r\n    public KafkaLogAppender(Properties props, String topic) {\r\n        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, KafkaAvroSerializer.class);\r\n        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, KafkaAvroSerializer.class);\r\n        this.producer = new KafkaProducer<String, IndexedRecord>(props);\r\n        this.topic = topic;\r\n    }\r\n\r\n    @Override\r\n    public void activateOptions() {\r\n        if (this.brokerList == null)\r\n            throw new ConfigException(\"The bootstrap servers property is required\");\r\n\r\n        if (this.schemaRegistryUrl == null)\r\n            throw new ConfigException(\"The schema registry url is required\");\r\n\r\n        if (this.topic == null)\r\n            throw new ConfigException(\"Topic is required\");\r\n\r\n        Properties props = new Properties();\r\n        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, this.brokerList);\r\n        props.put(\"schema.registry.url\", this.schemaRegistryUrl);\r\n        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, KafkaAvroSerializer.class);\r\n        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, KafkaAvroSerializer.class);\r\n\r\n        this.producer = new KafkaProducer<String, IndexedRecord>(props);\r\n    }\r\n\r\n    @Override\r\n    protected void append(LoggingEvent event) {\r\n        LogLine line = new LogLine();\r\n        line.setLine(event.getMessage() == null ? null : event.getMessage().toString());\r\n        line.setLogtypeid(logLevels.get(event.getLevel()));\r\n        line.setSource(event.getThreadName());\r\n\r\n        Map<CharSequence, Long> timings = new HashMap<CharSequence, Long>();\r\n        timings.put(\"emitted\", event.getTimeStamp());\r\n\r\n        line.setTimings(timings);\r\n\r\n        Map<CharSequence, CharSequence> tags = new HashMap<CharSequence, CharSequence>();\r\n        tags.put(\"logger.name\", event.getLoggerName());\r\n        tags.put(\"location\", event.getLocationInformation().fullInfo);\r\n\r\n        line.setTag(tags);\r\n\r\n        ProducerRecord<String, IndexedRecord> record = new ProducerRecord<String, IndexedRecord>(this.topic, line);\r\n        producer.send(record);\r\n    }\r\n\r\n    @Override\r\n    public void close() {\r\n        this.producer.close();\r\n    }\r\n\r\n    @Override\r\n    public boolean requiresLayout() {\r\n        return false;\r\n    }\r\n\r\n    public String getTopic() {\r\n        return topic;\r\n    }\r\n\r\n    public void setTopic(String topic) {\r\n        this.topic = topic;\r\n    }\r\n\r\n    public String getBrokerList() {\r\n        return brokerList;\r\n    }\r\n\r\n    public void setBrokerList(String brokerList) {\r\n        this.brokerList = brokerList;\r\n    }\r\n\r\n    public String getSchemaRegistryUrl() {\r\n        return schemaRegistryUrl;\r\n    }\r\n\r\n    public void setSchemaRegistryUrl(String schemaRegistryUrl) {\r\n        this.schemaRegistryUrl = schemaRegistryUrl;\r\n    }\r\n}\r\n```\r\nas recommended reporters to start with that we will want to start to evolve and work with others on too (maybe even) to help get some consensus around. \r\n\r\n```\r\n/**\r\n * KafkaLogAppender produces all log events to Kafka topic encoded as Avro.\r\n * Avro schema is defined as follows:\r\n * {\r\n *     \"type\": \"record\",\r\n *     \"name\": \"ly.stealth.avro.LogLine\",\r\n *     \"fields\": [\r\n *         {\r\n *             \"name\": \"line\",\r\n *             \"type\": [\"null\", \"string\"],\r\n *             \"default\": null\r\n *         },\r\n *         {\r\n *             \"name\": \"source\",\r\n *             \"type\": [\"null\", \"string\"],\r\n *             \"default\": null\r\n *         },\r\n *         {\r\n *             \"name\": \"tag\",\r\n *             \"type\": [\r\n *                 \"null\",\r\n *                 {\r\n *                     \"type\": \"map\",\r\n *                     \"values\": \"string\"\r\n *                 }\r\n *             ],\r\n *             \"default\": null\r\n *         },\r\n *         {\r\n *             \"name\": \"logtypeid\",\r\n *             \"type\": [\"null\", \"long\"],\r\n *             \"default\": null\r\n *         },\r\n *         {\r\n *             \"name\": \"timings\",\r\n *             \"type\": [\r\n *                 \"null\",\r\n *                 {\r\n *                     \"type\": \"map\",\r\n *                     \"values\": \"long\"\r\n *                 }\r\n *             ],\r\n *             \"default\": null\r\n *         }\r\n *     ]\r\n * }\r\n *\r\n * \"line\" field will contain the actual log message.\r\n * \"logtypeid\" will be the log level:\r\n *      1 - Trace\r\n *      2 - Debug\r\n *      3 - Info\r\n *      4 - Warn\r\n *      5 - Error\r\n *      6 - Fatal\r\n * \"source\" will be the thread name that logged the message.\r\n * \"tag\" will contain 2 entries:\r\n *      \"logger.name\" - logger name, e.g. the one that is used in call Logger.getLogger()\r\n *      \"location\" - event location information, e.g. source code line that logged the message\r\n * \"timings\" will contain 1 entry:\r\n *      \"emitted\" - timestamp in milliseconds when the message was sent\r\n */\r\n```\r\n\r\n\r\n"}}