Skip to main content
Top
Review meshIQ Products (G2.com)

Streaming Ingestion with TNT4J-Streams: Flume, Logstash, Beats, JMS, Kafka & MQTT

This article demonstrates how to ingest and normalize real-time events with TNT4J-Streams across Apache Flume (raw & parsed), Logstash (raw & parsed), Elastic Beats, JMS (text/map/object), Kafka, and MQTT—using JSON envelopes and stacked-parser patterns.

Apache Flume Raw Data

This sample shows how to stream activity events from redirected Apache Flume output raw data. Apache Flume output is configured to send raw output data as JSON to localhost:9595. This sample also shows how to use the stacked-parsers technique to extract log entry data from JSON envelope.

Sample files can be found in samples/apache-flume directory.

To install the plugin into Apache Flume and configure, see Installation of Apache Flume TNT4J-Streams Plugin.

Sample stream configuration:

<?xml version="1.0" encoding="utf-8"?>
<tnt-data-source
        xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
        xsi:noNamespaceSchemaLocation="https://meshiq-xsd.s3.amazonaws.com/streams/tnt-data-source.xsd">

    <parser name="AccessLogParserCommon" class="com.jkoolcloud.tnt4j.streams.custom.parsers.ApacheAccessLogParser">
        <property name="LogPattern" value="%h %l %u %t &quot;%r&quot; %s %b"/>

        <field name="Location" locator="1" locator-type="Index"/>
        <field name="UserName" locator="3" locator-type="Index"/>
        <field name="StartTime" locator="4" locator-type="Index" format="dd/MMM/yyyy:HH:mm:ss Z" datatype="DateTime" locale="en-US"/>
        <field name="EventType" value="SEND"/>
        <field name="EventName" locator="7" locator-type="Index"/>
        <field name="ResourceName" locator="8" locator-type="Index"/>
        <field name="CompCode" locator="12" locator-type="Index">
            <field-map source="100:206" target="SUCCESS" type="Range"/>
            <field-map source="300:308" target="WARNING" type="Range"/>
            <field-map source="400:417" target="ERROR" type="Range"/>
            <field-map source="500:511" target="ERROR" type="Range"/>
        </field>
        <field name="ReasonCode" locator="12" locator-type="Index"/>
        <field name="MsgValue" locator="13" locator-type="Index"/>
    </parser>

    <parser name="JSONEnvelopeParser" class="com.jkoolcloud.tnt4j.streams.parsers.ActivityJsonParser">
        <property name="ActivityDelim" value="EOL"/>

        <field name="MsgBody" locator="$.body" locator-type="Label">
            <parser-ref name="AccessLogParserCommon"/>
        </field>
        <field name="sinkName" locator="$.sinkName" locator-type="Label"/>
        <field name="chanelName" locator="$.chanelName" locator-type="Label"/>
        <field name="headers" locator="$.headers" locator-type="Label"/>
    </parser>

    <stream name="SampleFlumeStream" class="com.jkoolcloud.tnt4j.streams.inputs.CharacterStream">
        <property name="HaltIfNoParser" value="false"/>
        <property name="Port" value="9595"/>
        <parser-ref name="JSONEnvelopeParser"/>
    </stream>
</tnt-data-source>

Stream configuration states that CharacterStream referencing JSONEnvelopeParser shall be used.

CharacterStream starts a server socket on the port defined using the Port property. HaltIfNoParser property indicates that stream should skip unparseable entries.

JSONEnvelopeParser transforms the received JSON data package into a Map with entries MsgBody, sinkName, chanelName and headers. MsgBody entry value is passed to stacked parser named AccessLogParserCommon. ActivityDelim property indicates that every line in the parsed string represents a single JSON data package.

Details on AccessLogParserCommon (or ApacheAccessLogParser in general) can be found in the samples section Apache Access Log Single File and the parsers configuration section Apache Access Log Parser.

Stream stops only when a critical runtime error/exception occurs or the application gets terminated.

Apache Flume Parsed Data

This sample shows how to stream activity events from Apache Flume parsed log entry output data. Apache Flume output is configured to send parsed log entry data as JSON to localhost:9595.

Sample files can be found in the samples/apache-flume-parsed directory.

To install the plugin into Apache Flume and configure, see Installation of Apache Flume TNT4J-Streams Plugin.

The messages.json file contains a sample Apache Flume output JSON data package prepared using configuration of this sample. This sample JSON is for you to see and better understand the parsers mappings. Do not use it as Apache Flume input!

Sample stream configuration:

<?xml version="1.0" encoding="utf-8"?>
<tnt-data-source
        xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
        xsi:noNamespaceSchemaLocation="https://meshiq-xsd.s3.amazonaws.com/streams/tnt-data-source.xsd">

    <parser name="FlumeJSONParser" class="com.jkoolcloud.tnt4j.streams.parsers.ActivityJsonParser">
        <property name="ActivityDelim" value="EOL"/>

        <field name="Location" locator="$.headers.clientip" locator-type="Label"/>
        <field name="UserName" locator="$.headers.auth" locator-type="Label"/>
        <field name="StartTime" locator="$.headers.logtime" locator-type="Label" format="dd/MMM/yyyy:HH:mm:ss Z" datatype="DateTime" locale="en-US"/>
        <field name="EventType" value="SEND"/>
        <field name="EventName" locator="$.headers.method" locator-type="Label"/>
        <field name="ResourceName" locator="$.headers.param" locator-type="Label"/>
        <field name="CompCode" locator="$.headers.response" locator-type="Label">
            <field-map source="100:206" target="SUCCESS" type="Range"/>
            <field-map source="300:308" target="WARNING" type="Range"/>
            <field-map source="400:417" target="ERROR" type="Range"/>
            <field-map source="500:511" target="ERROR" type="Range"/>
        </field>
        <field name="ReasonCode" locator="$.headers.response" locator-type="Label"/>
        <field name="MsgValue" locator="$.headers.bytes" locator-type="Label"/>
        <field name="Message" locator="$.body" locator-type="Label"/>
        <field name="Tag" separator=",">
            <field-locator locator="$.sinkName" locator-type="Label"/>
            <field-locator locator="$.chanelName" locator-type="Label"/>
        </field>
    </parser>

    <stream name="SampleFlumeStream" class="com.jkoolcloud.tnt4j.streams.inputs.CharacterStream">
        <property name="Port" value="9595"/>
        <parser-ref name="FlumeJSONParser"/>
    </stream>
</tnt-data-source>

Stream configuration states that CharacterStream referencing FlumeJSONParser shall be used.

CharacterStream starts a server socket on the port defined using Port property.

FlumeJSONParser transforms the received JSON data package into Map entries. Note that some entries like headers in the map have an inner map as a value. Fields of such entries can be accessed by defining the field name using . as the field-hierarchy separator. ActivityDelim property indicates that every line in the parsed string represents a single JSON data package.

Details on AccessLogParserCommon (or ApacheAccessLogParser in general) can be found in the samples section Apache Access Log Single File and the parsers configuration section Apache Access Log Parser.

Logstash Raw Data

This sample shows how to stream activity events from redirected Logstash output raw data. Logstash output is configured to send raw output data as JSON to localhost:9595. This sample also shows how to use the stacked-parsers technique to extract log entry data from JSON envelope.

Sample files can be found in samples/logstash directory.

To set up the sample environment, see Using Logstash with TNT4J-Streams.

messages.json file contains sample Logstash output JSON data package prepared using configuration of this sample. This sample JSON is for you to see and better understand the parsers mappings. Do not use it as Logstash input!

Sample stream configuration:

<?xml version="1.0" encoding="utf-8"?>
<tnt-data-source
        xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
        xsi:noNamespaceSchemaLocation="https://meshiq-xsd.s3.amazonaws.com/streams/tnt-data-source.xsd">

    <parser name="AccessLogParserCommon" class="com.jkoolcloud.tnt4j.streams.custom.parsers.ApacheAccessLogParser">
        <property name="LogPattern" value="%h %l %u %t &quot;%r&quot; %s %b"/>

        <field name="Location" locator="1" locator-type="Index"/>
        <field name="UserName" locator="3" locator-type="Index"/>
        <field name="StartTime" locator="4" locator-type="Index" format="dd/MMM/yyyy:HH:mm:ss Z" datatype="DateTime" locale="en-US"/>
        <field name="EventType" value="SEND"/>
        <field name="EventName" locator="7" locator-type="Index"/>
        <field name="ResourceName" locator="8" locator-type="Index"/>
        <field name="CompCode" locator="12" locator-type="Index">
            <field-map source="100:206" target="SUCCESS" type="Range"/>
            <field-map source="300:308" target="WARNING" type="Range"/>
            <field-map source="400:417" target="ERROR" type="Range"/>
            <field-map source="500:511" target="ERROR" type="Range"/>
        </field>
        <field name="ReasonCode" locator="12" locator-type="Index"/>
        <field name="MsgValue" locator="13" locator-type="Index"/>
    </parser>

    <parser name="JSONEnvelopeParser" class="com.jkoolcloud.tnt4j.streams.parsers.ActivityJsonParser">
        <property name="ActivityDelim" value="EOL"/>

        <field name="MsgBody" locator="$.message" locator-type="Label">
            <parser-ref name="AccessLogParserCommon" tags="Normal server,Delayed server"/>
        </field>
        <field name="path" locator="$.path" locator-type="Label"/>
        <field name="Tag" locator="$.tags" locator-type="Label"/>
        <field name="host" locator="$.host" locator-type="Label"/>
    </parser>

    <stream name="SampleLogstashStream" class="com.jkoolcloud.tnt4j.streams.inputs.CharacterStream">
        <property name="HaltIfNoParser" value="false"/>
        <property name="RestartOnInputClose" value="true"/>
        <property name="Port" value="9595"/>
        <parser-ref name="JSONEnvelopeParser"/>
    </stream>
</tnt-data-source>

Stream configuration states that CharacterStream referencing JSONEnvelopeParser shall be used.

CharacterStream starts a server socket on the port defined using the Port property. HaltIfNoParser property indicates that the stream should skip unparseable entries.

JSONEnvelopeParser transforms the received JSON data package into a Map with entries MsgBody, path, Tag and host. MsgBody entry value is passed to stacked parser named AccessLogParserCommon. ActivityDelim property indicates that every line in the parsed string represents a single JSON data package.

Details on AccessLogParserCommon (or ApacheAccessLogParser in general) can be found in the samples section Apache Access Log Single File and the parsers configuration section Apache Access Log Parser.

Stream stops only when a critical runtime error/exception occurs or the application gets terminated.

Logstash Parsed Data

This sample shows how to stream activity events parsed by Logstash. Logstash Grok output plugin is configured to send parsed Apache Access log entry data as JSON to localhost:9595.

Sample files can be found in samples/logstash-parsed directory.

To set up the sample environment, see Using Logstash with TNT4J-Streams.

messages.json file contains sample Logstash output JSON data package prepared using configuration of this sample. This sample JSON is for you to see and better understand the parsers mappings. Do not use it as Logstash input!

Sample stream configuration:

<?xml version="1.0" encoding="utf-8"?>
<tnt-data-source
        xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
        xsi:noNamespaceSchemaLocation="https://meshiq-xsd.s3.amazonaws.com/streams/tnt-data-source.xsd">

    <parser name="LogstashJSONParser" class="com.jkoolcloud.tnt4j.streams.parsers.ActivityJsonParser">
        <property name="ActivityDelim" value="EOL"/>

        <field name="Location" locator="$.clientip" locator-type="Label"/>
        <field name="UserName" locator="$.auth" locator-type="Label"/>
        <field name="StartTime" locator="$.timestamp" locator-type="Label" format="dd/MMM/yyyy:HH:mm:ss Z" datatype="DateTime" locale="en-US"/>
        <field name="EventType" value="SEND"/>
        <field name="EventName" locator="$.verb" locator-type="Label"/>
        <field name="ResourceName" locator="$.request" locator-type="Label"/>
        <field name="CompCode" locator="$.response" locator-type="Label">
            <field-map source="100:206" target="SUCCESS" type="Range"/>
            <field-map source="300:308" target="WARNING" type="Range"/>
            <field-map source="400:417" target="ERROR" type="Range"/>
            <field-map source="500:511" target="ERROR" type="Range"/>
        </field>
        <field name="ReasonCode" locator="$.response" locator-type="Label"/>
        <field name="MsgValue" locator="$.bytes" locator-type="Label"/>
        <field name="Message" locator="$.message" locator-type="Label"/>
        <field name="Tag" locator="$.tags" locator-type="Label"/>
    </parser>

    <stream name="SampleLogstashStream" class="com.jkoolcloud.tnt4j.streams.inputs.CharacterStream">
        <property name="Port" value="9595"/>
        <parser-ref name="LogstashJSONParser"/>
    </stream>
</tnt-data-source>

Stream configuration states that CharacterStream referencing LogstashJSONParser shall be used.

CharacterStream starts a server socket on the port defined using the  Port property.

LogstashJSONParser transforms the received JSON data package to a Map data structure and the map entries to activity event fields using map entry key labels. The ActivityDelim property with value EOL indicates that every line in the parsed stream represents a single JSON data package.

Elastic Beats Provided Data

This sample shows how to stream activity events from Elastic Beats. 
The Elastic Beats environment must be configured to output data to a stream-started Logstash server host and port, e.g.  localhost:5044.

Sample files can be found in samples/elastic-beats directory.

The dashboards directory contains an exported jKool dashboard dedicated to visualizing data for this sample. You can import it into your jKool repository.

To set up Elastic Beats environment, see Configuring Elastic Beats.

sampleMsg.json file contains a sample Elastic Beats provided Logstash message as JSON data prepared using configuration of this sample. This sample JSON is for you to see and better understand the parsers mappings. Do not use it as Logstash input!

Sample stream configuration:

<?xml version="1.0" encoding="utf-8"?>
<tnt-data-source
        xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
        xsi:noNamespaceSchemaLocation="https://meshiq-xsd.s3.amazonaws.com/streams/tnt-data-source.xsd">

    <parser name="BeatsMessageParser" class="com.jkoolcloud.tnt4j.streams.parsers.ActivityMapParser">
        <field name="EventType" value="EVENT"/>

        <field name="StartTime">
            <field-locator locator="@timestamp" locator-type="Label" datatype="DateTime" format="yyyy-MM-dd'T'HH:mm:ss.SSSX">
            </field-locator>
        </field>

        <field name="EventName" formattingPattern="{0} {1}">
            <field-locator locator="metricset.module" locator-type="Label"/>
            <field-locator locator="metricset.name" locator-type="Label"/>
        </field>

        <field name="all" locator="*" locator-type="Label"/>
    </parser>

    <stream name="SampleBeatsStream" class="com.jkoolcloud.tnt4j.streams.inputs.ElasticBeatsStream">
        <property name="HaltIfNoParser" value="false"/>
        <property name="RestartOnInputClose" value="true"/>

        <!--Binding host-->
        <!--<property name="Host" value="localhost"/>-->
        <property name="Port" value="5044"/>
        <!--Worker properties-->
        <!--<property name="Timeout" value="30"/>-->
        <!--<property name="ThreadCount" value="1"/>-->

        <!--SSL properties-->
        <!--<property name="SSLCertificateFilePath" value="/etc/pki/client/cert.key"/>-->
        <!--<property name="SSLKeyFilePath" value="/etc/pki/client/cert.pem"/>-->
        <!--<property name="PassPhrase" value="pass"/>-->

        <parser-ref name="BeatsMessageParser"/>
    </stream>
</tnt-data-source>

Stream configuration states that ElasticBeatsStream referencing BeatsMessageParser shall be used.

ElasticBeatsStream starts Logstash server on port defined using Port property. You can also define host for server to bind by using Host property.

BeatsMessageParser takes the map data structure provided by the stream and the map entries to activity event fields using map entry key labels. Since there is no particular set of predefined fields for Elastic Beats data, in this sample we map them directly into activity entity (EVENT) properties using the locator * to include them all in jKool.

JMS Text Message

This sample shows how to stream activity events received over JMS transport as text messages. This sample also shows how to use the stacked-parsers technique to extract message payload data.

Sample files can be found in samples/jms-textmessage directory.

In run.bat/run.sh file set variable JMS_IMPL_LIBPATH value to reference JMS implementation (ActiveMQ, Solace, RabbitMQ, etc.) libraries used by your environment.

Sample stream configuration:

<?xml version="1.0" encoding="utf-8"?>
<tnt-data-source
        xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
        xsi:noNamespaceSchemaLocation="https://meshiq-xsd.s3.amazonaws.com/streams/tnt-data-source.xsd">

    <parser name="AccessLogParserCommon" class="com.jkoolcloud.tnt4j.streams.custom.parsers.ApacheAccessLogParser">
        <property name="LogPattern" value="%h %l %u %t &quot;%r&quot; %s %b"/>

        <field name="Location" locator="1" locator-type="Index"/>
        <field name="UserName" locator="3" locator-type="Index"/>
        <field name="StartTime" locator="4" locator-type="Index" format="dd/MMM/yyyy:HH:mm:ss Z" datatype="DateTime" locale="en-US"/>
        <field name="EventType" value="SEND"/>
        <field name="EventName" locator="7" locator-type="Index"/>
        <field name="ResourceName" locator="8" locator-type="Index"/>
        <field name="CompCode" locator="12" locator-type="Index">
            <field-map source="100:206" target="SUCCESS" type="Range"/>
            <field-map source="300:308" target="WARNING" type="Range"/>
            <field-map source="400:417" target="ERROR" type="Range"/>
            <field-map source="500:511" target="ERROR" type="Range"/>
        </field>
        <field name="ReasonCode" locator="12" locator-type="Index"/>
        <field name="MsgValue" locator="13" locator-type="Index"/>
    </parser>

    <parser name="SampleJMSParser" class="com.jkoolcloud.tnt4j.streams.parsers.ActivityJMSMessageParser">
        <field name="Transport" locator="ActivityTransport" locator-type="Label"/>
        <field name="MsgBody" locator="ActivityData" locator-type="Label">
            <parser-ref name="AccessLogParserCommon"/>
        </field>

        <!-- mapping common message metadata fields one by one -->
        <field name="Correlator" locator="MsgMetadata.Correlator" locator-type="Label"/>
        <field name="CorrelatorBytes" locator="MsgMetadata.CorrelatorBytes" locator-type="Label"/>
        <field name="DeliveryMode" locator="MsgMetadata.DeliveryMode" locator-type="Label"/>
        <field name="Destination" locator="MsgMetadata.Destination" locator-type="Label"/>
        <field name="Expiration" locator="MsgMetadata.Expiration" locator-type="Label"/>
        <field name="MessageId" locator="MsgMetadata.MessageId" locator-type="Label"/>
        <field name="Priority" locator="MsgMetadata.Priority" locator-type="Label"/>
        <field name="Redelivered" locator="MsgMetadata.Redelivered" locator-type="Label"/>
        <field name="ReplyTo" locator="MsgMetadata.ReplyTo" locator-type="Label"/>
        <field name="Timestamp" locator="MsgMetadata.Timestamp" locator-type="Label"/>
        <field name="Type" locator="MsgMetadata.Type" locator-type="Label"/>
        <!-- automatically puts all unmapped message metadata map entries as custom activity properties -->
        <field name="AllRestMsgMetadataProps" locator="MsgMetadata.#" locator-type="Label"/>
        <!-- automatically puts all resolved custom message properties map entries as custom activity properties -->
        <field name="CustomMsgProps" locator="MsgMetadata.CustomMsgProps" locator-type="Label"/>
        <!-- or mapping (some) custom message properties one by one-->
        <!--<field name="CustomProp1" locator="MsgMetadata.CustomMsgProps.Property1" locator-type="Label"/>-->
        <!--<field name="CustomProp2" locator="MsgMetadata.CustomMsgProps.Property2" locator-type="Label"/>-->
        <!--<field name="CustomProp3" locator="MsgMetadata.CustomMsgProps.Property3" locator-type="Label"/>-->
        <!-- and all what is left unmapped, map automatically -->
        <!--<field name="AllRestCustomMsgProps" locator="MsgMetadata.CustomMsgProps.#" locator-type="Label"/>-->

        <!-- mapping message metadata fields as map entries-->
        <!-- automatically puts all resolved map entries as custom activity properties -->
        <!--<field name="MsgMetadata" locator="MsgMetadata" locator-type="Label"/>-->
    </parser>

    <stream name="SampleJMStream" class="com.jkoolcloud.tnt4j.streams.inputs.JMSStream">
        <property name="HaltIfNoParser" value="false"/>
        <property name="java.naming.provider.url" value="tcp://localhost:61616"/>
        <property name="java.naming.factory.initial" value="org.apache.activemq.jndi.ActiveMQInitialContextFactory"/>
        <!--<property name="java.naming.security.username" value="[YOUR_USERNAME]"/>-->
        <!--<property name="java.naming.security.principal" value="[YOUR_PRINCIPAL]"/>-->
        <!--<property name="java.naming.security.credentials" value="[YOUR_PASSWORD]"/>-->
        <!--<property name="Queue" value="queue.SampleJMSQueue"/>-->
        <!--<property name="Queue" value="dynamicQueues/queue.SampleJMSQueue"/>-->
        <property name="Topic" value="TestTopic"/>
        <!--<property name="Topic" value="dynamicTopics/TestTopic"/>-->
        <property name="JMSConnFactory" value="/jms/cf/another"/>

        <parser-ref name="SampleJMSParser"/>
    </stream>
</tnt-data-source>

Stream configuration states that JMSStream referencing SampleJMSParser shall be used.

JMSStream connects to the server defined using the java.naming.provider.url property, and takes messages from the topic defined by the Topic property. To define wanted queue use Queue property. HaltIfNoParser property indicates that stream should skip unparseable entries. java.naming.factory.initial property defines that ActiveMQ shall be used. Stream puts the received message data into a map and passes it to the parser.

SampleJMSParser maps metadata to activity event data. the ActivityData entry value is passed to the stacked parser named AccessLogParserCommon.

Details on AccessLogParserCommon (or ApacheAccessLogParser in general) can be found in the samples section Apache Access Log Single File and the parser configuration section Apache Access Log Parser.

To parse some other data instead of Apache Access Log, replace AccessLogParserCommon with a parser that complies with your data format.

Stream stops only when a critical runtime error/exception occurs or the application gets terminated.

JMS Map Message

This sample shows how to stream activity events received over JMS transport as map messages.

Sample files can be found in samples/jms-mapmessage directory.

In run.bat/run.sh file set variable JMS_IMPL_LIBPATH value to reference JMS implementation (ActiveMQ, Solace, RabbitMQ, etc.) libraries used by your environment.

Sample stream configuration:

<?xml version="1.0" encoding="utf-8"?>
<tnt-data-source
        xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
        xsi:noNamespaceSchemaLocation="https://meshiq-xsd.s3.amazonaws.com/streams/tnt-data-source.xsd">

    <parser name="SampleJMSParser" class="com.jkoolcloud.tnt4j.streams.parsers.ActivityJMSMessageParser">
        <field name="Transport" locator="ActivityTransport" locator-type="Label"/>
        <field name="Location" locator="clientip" locator-type="Label"/>
        <field name="UserName" locator="auth" locator-type="Label"/>
        <field name="StartTime" locator="timestamp" locator-type="Label" format="dd/MMM/yyyy:HH:mm:ss Z" datatype="DateTime" locale="en-US"/>
        <field name="EventType" value="SEND"/>
        <field name="EventName" locator="verb" locator-type="Label"/>
        <field name="ResourceName" locator="request" locator-type="Label"/>
        <field name="CompCode" locator="response" locator-type="Label">
            <field-map source="100:206" target="SUCCESS" type="Range"/>
            <field-map source="300:308" target="WARNING" type="Range"/>
            <field-map source="400:417" target="ERROR" type="Range"/>
            <field-map source="500:511" target="ERROR" type="Range"/>
        </field>
        <field name="ReasonCode" locator="response" locator-type="Label"/>
        <field name="MsgValue" locator="bytes" locator-type="Label"/>

        <!-- mapping common message metadata fields one by one -->
        <field name="Correlator" locator="MsgMetadata.Correlator" locator-type="Label"/>
        <field name="CorrelatorBytes" locator="MsgMetadata.CorrelatorBytes" locator-type="Label"/>
        <field name="DeliveryMode" locator="MsgMetadata.DeliveryMode" locator-type="Label"/>
        <field name="Destination" locator="MsgMetadata.Destination" locator-type="Label"/>
        <field name="Expiration" locator="MsgMetadata.Expiration" locator-type="Label"/>
        <field name="MessageId" locator="MsgMetadata.MessageId" locator-type="Label"/>
        <field name="Priority" locator="MsgMetadata.Priority" locator-type="Label"/>
        <field name="Redelivered" locator="MsgMetadata.Redelivered" locator-type="Label"/>
        <field name="ReplyTo" locator="MsgMetadata.ReplyTo" locator-type="Label"/>
        <field name="Timestamp" locator="MsgMetadata.Timestamp" locator-type="Label"/>
        <field name="Type" locator="MsgMetadata.Type" locator-type="Label"/>
        <!-- automatically puts all unmapped message metadata map entries as custom activity properties -->
        <field name="AllRestMsgMetadataProps" locator="MsgMetadata.#" locator-type="Label"/>
        <!-- automatically puts all resolved custom message properties map entries as custom activity properties -->
        <field name="CustomMsgProps" locator="MsgMetadata.CustomMsgProps" locator-type="Label"/>
        <!-- or mapping (some) custom message properties one by one-->
        <!--<field name="CustomProp1" locator="MsgMetadata.CustomMsgProps.Property1" locator-type="Label"/>-->
        <!--<field name="CustomProp2" locator="MsgMetadata.CustomMsgProps.Property2" locator-type="Label"/>-->
        <!--<field name="CustomProp3" locator="MsgMetadata.CustomMsgProps.Property3" locator-type="Label"/>-->
        <!-- and all what is left unmapped, map automatically -->
        <!--<field name="AllRestCustomMsgProps" locator="MsgMetadata.CustomMsgProps.#" locator-type="Label"/>-->

        <!-- mapping message metadata fields as map entries-->
        <!-- automatically puts all resolved map entries as custom activity properties -->
        <!--<field name="MsgMetadata" locator="MsgMetadata" locator-type="Label"/>-->
    </parser>

    <stream name="SampleJMStream" class="com.jkoolcloud.tnt4j.streams.inputs.JMSStream">
        <property name="HaltIfNoParser" value="false"/>
        <property name="java.naming.provider.url" value="tcp://localhost:61616"/>
        <property name="java.naming.factory.initial" value="org.apache.activemq.jndi.ActiveMQInitialContextFactory"/>
        <!--<property name="java.naming.security.username" value="[YOUR_USERNAME]"/>-->
        <!--<property name="java.naming.security.principal" value="[YOUR_PRINCIPAL]"/>-->
        <!--<property name="java.naming.security.credentials" value="[YOUR_PASSWORD]"/>-->
        <!--<property name="Queue" value="queue.SampleJMSQueue"/>-->
        <!--<property name="Queue" value="dynamicQueues/queue.SampleJMSQueue"/>-->
        <property name="Topic" value="TestTopic"/>
        <!--<property name="Topic" value="dynamicTopics/TestTopic"/>-->
        <property name="JMSConnFactory" value="/jms/cf/another"/>

        <parser-ref name="SampleJMSParser"/>
    </stream>
</tnt-data-source>

Stream configuration states that JMSStream referencing SampleJMSParser shall be used.

JMSStream connects to the server defined using the java.naming.provider.url property, and takes messages from the topic defined by the Topic property. To define the desired queue, use the Queue property. HaltIfNoParser property indicates that stream should skip unparseable entries. java.naming.factory.initial property defines that ActiveMQ shall be used. Stream puts the received message data into a map and passes it to the parser.

SampleJMSParser maps activity event data from a JMS map message using map entry key labels.

Stream stops only when a critical runtime error/exception occurs or the application gets terminated.

JMS Object Message

This sample shows how to stream activity events received over JMS transport as serializable object messages. This sample also shows how to use the stacked-parsers technique to extract message payload data.

Sample files can be found in samples/jms-objectmessage directory.

In run.bat/run.sh file set variable JMS_IMPL_LIBPATH value to reference JMS implementation (ActiveMQ, Solace, RabbitMQ, etc.) libraries used by your environment.

Sample stream configuration:

<?xml version="1.0" encoding="utf-8"?>
<tnt-data-source
        xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
        xsi:noNamespaceSchemaLocation="https://meshiq-xsd.s3.amazonaws.com/streams/tnt-data-source.xsd">

    <parser name="SampleObjectParser" class="com.jkoolcloud.tnt4j.streams.parsers.ActivityJavaObjectParser">
        <field name="Location" locator="clientip" locator-type="Label"/>
        <field name="UserName" locator="auth" locator-type="Label"/>
        <field name="StartTime" locator="timestamp" locator-type="Label" format="dd/MMM/yyyy:HH:mm:ss Z" datatype="DateTime" locale="en-US"/>
        <field name="EventType" value="SEND"/>
        <field name="EventName" locator="verb" locator-type="Label"/>
        <field name="ResourceName" locator="request" locator-type="Label"/>
        <field name="CompCode" locator="response" locator-type="Label">
            <field-map source="100:206" target="SUCCESS" type="Range"/>
            <field-map source="300:308" target="WARNING" type="Range"/>
            <field-map source="400:417" target="ERROR" type="Range"/>
            <field-map source="500:511" target="ERROR" type="Range"/>
        </field>
        <field name="ReasonCode" locator="response" locator-type="Label"/>
        <field name="MsgValue" locator="bytes" locator-type="Label"/>
    </parser>

    <parser name="SampleJMSParser" class="com.jkoolcloud.tnt4j.streams.parsers.ActivityJMSMessageParser">
        <field name="Transport" locator="ActivityTransport" locator-type="Label"/>
        <field name="MsgBody" locator="ActivityData" locator-type="Label">
            <parser-ref name="SampleObjectParser"/>
        </field>

        <!-- mapping common message metadata fields one by one -->
        <field name="Correlator" locator="MsgMetadata.Correlator" locator-type="Label"/>
        <field name="CorrelatorBytes" locator="MsgMetadata.CorrelatorBytes" locator-type="Label"/>
        <field name="DeliveryMode" locator="MsgMetadata.DeliveryMode" locator-type="Label"/>
        <field name="Destination" locator="MsgMetadata.Destination" locator-type="Label"/>
        <field name="Expiration" locator="MsgMetadata.Expiration" locator-type="Label"/>
        <field name="MessageId" locator="MsgMetadata.MessageId" locator-type="Label"/>
        <field name="Priority" locator="MsgMetadata.Priority" locator-type="Label"/>
        <field name="Redelivered" locator="MsgMetadata.Redelivered" locator-type="Label"/>
        <field name="ReplyTo" locator="MsgMetadata.ReplyTo" locator-type="Label"/>
        <field name="Timestamp" locator="MsgMetadata.Timestamp" locator-type="Label"/>
        <field name="Type" locator="MsgMetadata.Type" locator-type="Label"/>
        <!-- automatically puts all unmapped message metadata map entries as custom activity properties -->
        <field name="AllRestMsgMetadataProps" locator="MsgMetadata.#" locator-type="Label"/>
        <!-- automatically puts all resolved custom message properties map entries as custom activity properties -->
        <field name="CustomMsgProps" locator="MsgMetadata.CustomMsgProps" locator-type="Label"/>
        <!-- or mapping (some) custom message properties one by one-->
        <!--<field name="CustomProp1" locator="MsgMetadata.CustomMsgProps.Property1" locator-type="Label"/>-->
        <!--<field name="CustomProp2" locator="MsgMetadata.CustomMsgProps.Property2" locator-type="Label"/>-->
        <!--<field name="CustomProp3" locator="MsgMetadata.CustomMsgProps.Property3" locator-type="Label"/>-->
        <!-- and all what is left unmapped, map automatically -->
        <!--<field name="AllRestCustomMsgProps" locator="MsgMetadata.CustomMsgProps.#" locator-type="Label"/>-->

        <!-- mapping message metadata fields as map entries-->
        <!-- automatically puts all resolved map entries as custom activity properties -->
        <!--<field name="MsgMetadata" locator="MsgMetadata" locator-type="Label"/>-->
    </parser>

    <stream name="SampleJMStream" class="com.jkoolcloud.tnt4j.streams.inputs.JMSStream">
        <property name="HaltIfNoParser" value="false"/>
        <property name="java.naming.provider.url" value="tcp://localhost:61616"/>
        <property name="java.naming.factory.initial" value="org.apache.activemq.jndi.ActiveMQInitialContextFactory"/>
        <!--<property name="java.naming.security.username" value="[YOUR_USERNAME]"/>-->
        <!--<property name="java.naming.security.principal" value="[YOUR_PRINCIPAL]"/>-->
        <!--<property name="java.naming.security.credentials" value="[YOUR_PASSWORD]"/>-->
        <!--<property name="Queue" value="queue.SampleJMSQueue"/>-->
        <!--<property name="Queue" value="dynamicQueues/queue.SampleJMSQueue"/>-->
        <property name="Topic" value="TestTopic"/>
        <!--<property name="Topic" value="dynamicTopics/TestTopic"/>-->
        <property name="JMSConnFactory" value="/jms/cf/another"/>

        <parser-ref name="SampleJMSParser"/>
    </stream>
</tnt-data-source>

Stream configuration states that JMSStream referencing SampleJMSParser shall be used.

JMSStream connects to the server defined using the java.naming.provider.url property, and takes messages from the topic defined by the Topic property. To define wanted queue use Queue property. HaltIfNoParser property indicates that stream should skip unparseable entries. java.naming.factory.initial property defines that ActiveMQ shall be used. Stream puts the received message data into a map and passes it to the parser.

SampleJMSParser maps metadata to activity event data. ActivityData entry value is passed to stacked parser named SampleObjectParser.

SampleObjectParser can map activity-event field values from the serialized object's declared fields using field names as labels.

Stream stops only when a critical runtime error/exception occurs or the application gets terminated.

ActiveMQ Artemis Interceptors

See the Artemis Interceptors Sample for details.

Kafka Consumer Stream

This sample shows how to stream activity events received over Apache Kafka transport as messages. This sample also shows how to use the stacked-parsers technique to extract message payload data.

Stream runs as an Apache Kafka consumer.

Sample files can be found in samples/kafka-client directory.

Sample stream configuration:

<?xml version="1.0" encoding="utf-8"?>
<tnt-data-source
        xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
        xsi:noNamespaceSchemaLocation="https://meshiq-xsd.s3.amazonaws.com/streams/tnt-data-source.xsd">

    <parser name="AccessLogParserCommon" class="com.jkoolcloud.tnt4j.streams.custom.parsers.ApacheAccessLogParser">
        <property name="LogPattern" value="%h %l %u %t &quot;%r&quot; %s %b"/>

        <field name="Location" locator="1" locator-type="Index"/>
        <field name="UserName" locator="3" locator-type="Index"/>
        <field name="StartTime" locator="4" locator-type="Index" format="dd/MMM/yyyy:HH:mm:ss Z" datatype="DateTime" locale="en-US"/>
        <field name="EventType" value="SEND"/>
        <field name="EventName" locator="7" locator-type="Index"/>
        <field name="ResourceName" locator="8" locator-type="Index"/>
        <field name="CompCode" locator="12" locator-type="Index">
            <field-map source="100:206" target="SUCCESS" type="Range"/>
            <field-map source="300:308" target="WARNING" type="Range"/>
            <field-map source="400:417" target="ERROR" type="Range"/>
            <field-map source="500:511" target="ERROR" type="Range"/>
        </field>
        <field name="ReasonCode" locator="12" locator-type="Index"/>
        <field name="MsgValue" locator="13" locator-type="Index"/>
    </parser>

    <parser name="KafkaMessageParser" class="com.jkoolcloud.tnt4j.streams.parsers.KafkaConsumerRecordParser">
        <field name="Topic" locator="topic" locator-type="Label"/>
        <field name="Partition" locator="partition" locator-type="Label" datatype="Number"/>
        <field name="Offset" locator="offset" locator-type="Label" datatype="Number"/>
        <field name="Timestamp" locator="timestamp" locator-type="Label" datatype="Timestamp"/>
        <field name="TimestampType" locator="timestampType" locator-type="Label"/>
        <field name="KeySize" locator="serializedKeySize" locator-type="Label" datatype="Number"/>
        <field name="ValueSize" locator="serializedValueSize" locator-type="Label" datatype="Number"/>
        <field name="Key" locator="key" locator-type="Label"/>
        <field name="MsgBody" locator="value" locator-type="Label">
            <parser-ref name="AccessLogParserCommon"/>
        </field>
    </parser>

    <stream name="SampleKafkaClientStream" class="com.jkoolcloud.tnt4j.streams.inputs.KafkaConsumerStream">
        <property name="HaltIfNoParser" value="false"/>
        <property name="RetryStateCheck" value="3"/>
        <property name="RetryInterval" value="5"/>
        <property name="Topic" value="TNT4JStreams"/>
        <property name="PollTimeoutMs" value="3000"/>
        <!--        <property name="TopicBindingStrategy" value="ASSIGN"/>-->
        <!--        <property name="PartitionsUpdatePeriodSec" value="60"/>-->

        <!-- Kafka consumer properties -->
        <property name="bootstrap.servers" value="localhost:9092"/>
        <property name="group.id" value="0"/>
        <property name="key.deserializer" value="org.apache.kafka.common.serialization.StringDeserializer"/>
        <property name="value.deserializer" value="org.apache.kafka.common.serialization.StringDeserializer"/>
        <property name="enable.auto.commit" value="true"/>
        <property name="auto.commit.interval.ms" value="1000"/>
        <property name="session.timeout.ms" value="30000"/>
        <property name="client.id" value="tnt4j-streams-kafka-consumer-stream"/>

        <parser-ref name="KafkaMessageParser"/>
    </stream>
</tnt-data-source>

Stream configuration states that SampleKafkaClientStream referencing KafkaMessageParser shall be used.

SampleKafkaClientStream connects to the server defined using the bootstrap.servers property, and takes messages from topic defined Topic property. HaltIfNoParser property indicates that stream should skip unparseable entries. Stream puts the received message data into a map and passes it to the parser.

See Apache Kafka Consumer configuration for details.

KafkaMessageParser maps metadata to activity event data. ActivityData entry value is passed to stacked parser named AccessLogParserCommon.

Details on AccessLogParserCommon (or ApacheAccessLogParser in general) can be found in the samples section Apache Access Log Single File and the parsers configuration section Apache Access Log Parser.

To parse some other data instead of Apache Access Log, replace AccessLogParserCommon with a parser that complies with your data format.

Stream stops only when a critical runtime error/exception occurs or the application gets terminated.

MQTT

This sample shows how to stream activity events received over MQTT transport as MQTT messages. This sample also shows how to use The stacked-parsers technique to extract message payload data.

Sample files can be found in samples/mqtt directory.

Sample stream configuration:

<?xml version="1.0" encoding="utf-8"?>
<tnt-data-source
        xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
        xsi:noNamespaceSchemaLocation="https://meshiq-xsd.s3.amazonaws.com/streams/tnt-data-source.xsd">

    <parser name="AccessLogParserCommon" class="com.jkoolcloud.tnt4j.streams.custom.parsers.ApacheAccessLogParser">
        <property name="LogPattern" value="%h %l %u %t &quot;%r&quot; %s %b"/>

        <field name="Location" locator="1" locator-type="Index"/>
        <field name="UserName" locator="3" locator-type="Index"/>
        <field name="StartTime" locator="4" locator-type="Index" format="dd/MMM/yyyy:HH:mm:ss Z" datatype="DateTime" locale="en-US"/>
        <field name="EventType" value="SEND"/>
        <field name="EventName" locator="7" locator-type="Index"/>
        <field name="ResourceName" locator="8" locator-type="Index"/>
        <field name="CompCode" locator="12" locator-type="Index">
            <field-map source="100:206" target="SUCCESS" type="Range"/>
            <field-map source="300:308" target="WARNING" type="Range"/>
            <field-map source="400:417" target="ERROR" type="Range"/>
            <field-map source="500:511" target="ERROR" type="Range"/>
        </field>
        <field name="ReasonCode" locator="12" locator-type="Index"/>
        <field name="MsgValue" locator="13" locator-type="Index"/>
    </parser>

    <parser name="MqttMessageParser" class="com.jkoolcloud.tnt4j.streams.parsers.ActivityMapParser">
        <field name="Topic" locator="ActivityTopic" locator-type="Label"/>
        <field name="Transport" locator="ActivityTransport" locator-type="Label"/>
        <field name="MsgBody" locator="ActivityData" locator-type="Label">
            <parser-ref name="AccessLogParserCommon"/>
        </field>
    </parser>

    <stream name="SampleMQTTStream" class="com.jkoolcloud.tnt4j.streams.inputs.MqttStream">
        <property name="HaltIfNoParser" value="false"/>
        <property name="ServerURI" value="tcp://localhost:1883"/>
        <property name="TopicString" value="TNT4JStreams"/>
        <!--<property name="UserName" value="someUser"/>-->
        <!--<property name="Password" value="somePassword"/>-->
        <!--<property name="UseSSL" value="true"/>-->
        <!--<property name="Keystore" value="path_to_keystore_file"/>-->
        <!--<property name="KeystorePass" value="somePassword"/>-->

        <parser-ref name="MqttMessageParser"/>
    </stream>
</tnt-data-source>

Stream configuration states that MqttStream referencing MqttMessageParser shall be used.

MqttStream connects to the server defined using ServerURI property, and takes messages from topic defined TopicString property. HaltIfNoParser property indicates that stream should skip unparseable entries. Stream puts the received message data into a map and passes it to the parser.

MqttMessageParser maps metadata to activity event data. the ActivityData entry value is passed to the stacked parser named AccessLogParserCommon.

Details on AccessLogParserCommon (or ApacheAccessLogParser in general) can be found in the samples section Apache Access Log Single File and the parsers configuration section Apache Access Log Parser.

To parse some other data instead of Apache Access Log, replace AccessLogParserCommon with a parser that complies with your data format.

Stream stops only when a critical runtime error/exception occurs or the application gets terminated.