Socket Input Adapter

[Former Member]

Hi all,

I am trying to get streaming data with socket csv input adapter. The streaming data is machine generated event logs and does not include "\n" character. So, I could not get data into ESP.

When I try this with a simple java socket server and send string without "\n", ESP does not insert any data into IputStream. When I append "\n" to the string, then ESP insert streaming data into InputStream.

Do you have any idea about getting streaming data which does not include "\n".

Thanks and regards,

Bulut

[Former Member · Accepted solution]

Hi Bulut,

Some of the adapters supplied with ESP are built with the toolkit.  These adapters consist of modules as illustrated here:

Event Stream Processor Adapter Toolkit

If you look at the CSV Socket Input Adapter's configuration file (%ESP_HOME%\adapters\framework\instances\socket_csv_input\adapter_config.xml), you will see that it consists of four modules (an input transport, two formatters and an ESP connector):

SocketInputTransporter -> StreamToStringFormatter -> CsvStringToEspFormatter -> EspPublisher

After the socket input transporter reads a stream of TCP data, the StreamToStringFormatter needs to somehow identify where one line begins and the next ends so that it can begin to parse the data into columns that can then be published to ESP.

You can see what another user did here with their Socket Server:

Thanks,

Neal

[Former Member · Accepted solution]

Hi Neal,

Thanks for your answer. I have been looking the details of the adapter toolkit especially streaming input example. So, what I understand from your answer is that existing I should modify the adapter with adapter toolkit. Do you know more easier way to get streaming data which does not include newline ("\n") character?

Thanks and regards,

Bulut

[Former Member · Accepted solution]

Hello,

The source code for the CSV Socket Input Adapter is not publicly available so you would have to write you own from scratch.

Even if you wrote your own, how would you know where one line of data began and the next ended?  ESP deals with messages.  For the CSV Socket Input Adapter:

  1 line of data = 1 message

I don't understand how your data is structured such that as it was being read from the TCP socket server, how would the data be parsed into a single message to be published into ESP?

Thanks,

Neal

[Former Member · Accepted solution]

Yes, you are right, 1 line of data is 1 message. But, I am trying to parse a log data in Common Event Format (CEF). As you can see in the attached file, single CEF message is start with "CEF" and CEF messages are comming without a newline char. So, what my opinion is that read the message from tcp socket in length of packet buffer size. So, 1 line of data is the data in tcp packet with buffer size. Then, I parse the incoming messages in a FlexOperator.

I did the parsing CEF events in a file like attached file, but now I am trying same thing with tcp socket.

[Former Member · Accepted solution]

Sorry, I could not attached a file. You can see similar CEF data below:

CEF:0|Check Point|VPN-1 & FireWall-1||drop|drop|Low| eventId=111111111 msg=Address spoofing proto=TCP categorySignificance=/Informational/Warning categoryBehavior=/Access categoryTechnique=/Traffic Anomaly/Network Layer/Spoof categoryDeviceGroup=/Firewall categoryOutcome=/Failure categoryObject=/Host/Application/Service art=1111111111111 cat=SecurityLog act=drop rt=1111111111111 deviceDirection=0 src=1.22.333.444 sourceZoneURI=/All Zones/ArcSight System/Public Address Space Zones/APNIC/1.22.333.444-1.22.333.444 (APNIC) spt=637 dst=1.22.333.444 destinationZoneURI=/All Zones/ArcSight System/Public Address Space Zones/APNIC/1.22.333.444-1.22.333.444 (APNIC) dpt=500 destinationServiceName=ISAKMP cs1= &  cs3=Mg7-MM cs6=fwgtout-32992011tufin cs1Label=rule & Rule Name cs2Label=UFP category cs3Label=Manager cs4Label=Rule UID cs5Label=Total bytes cs6Label=Policy Name cn1Label=Elapsed Time in Seconds cn2Label=icmp_type cn3Label=icmp_code deviceCustomDate1Label=Local Time c6a2Label=Source IPv6 Address c6a3Label=Destination IPv6 Address ahost=1.22.333.444 agt=1.22.333.444 agentZoneURI=/All Zones/ArcSight System/Private Address Space Zones/RFC1918: 1.22.333.444-1.22.333.444 av=1.22.333.4444.0 atz=Europe/Istanbul aid=flL+iYoBABCAO6bMvUEZYZ\=\= at=checkpointfirewall_ad_opsec dvc=1.22.333.444 deviceZoneURI=/All Zones/ArcSight System/Private Address Space Zones/RFC1918: 1.22.333.444-1.22.333.444 dtz=Europe/Istanbul deviceInboundInterface=eth-s1p1c0 _cefVer=0.1 ad.has__accounting=0CEF:0|Check Point|VPN-1 & FireWall-1||drop|drop|High| eventId=111111111 msg=Address spoofing proto=UDP categorySignificance=/Informational/Warning categoryBehavior=/Access categoryTechnique=/Traffic Anomaly/Network Layer/Spoof categoryDeviceGroup=/Firewall categoryOutcome=/Failure categoryObject=/Host/Application/Service art=1111111111111 cat=SecurityLog act=drop rt=1111111111111 deviceDirection=0 shost=1-65-32-002.static.netvigator.com src=1.22.333.444 sourceZoneURI=/All Zones/ArcSight System/Public Address Space Zones/APNIC/1.22.333.444-1.22.333.444 (APNIC) spt=500 dhost=2-55-32-011.static.netvigator.com dst=1.22.333.444 destinationZoneURI=/All Zones/ArcSight System/Public Address Space Zones/APNIC/1.22.333.444-1.22.333.444 (APNIC) dpt=500 destinationServiceName=ISAKMP cs1= &  cs3=Mg1-NG cs6=fwgtout-30092011tufin cs1Label=rule & Rule Name cs2Label=UFP category cs3Label=Manager cs4Label=Rule UID cs5Label=Total bytes cs6Label=Policy Name cn1Label=Elapsed Time in Seconds cn2Label=icmp_type cn3Label=icmp_code deviceCustomDate1Label=Local Time c6a2Label=Source IPv6 Address c6a3Label=Destination IPv6 Address ahost=1.22.333.444 agt=1.22.333.444 agentZoneURI=/All Zones/ArcSight System/Private Address Space Zones/RFC1918: 1.22.333.444-1.22.333.444 av=1.22.333.4444.0 atz=Europe/Istanbul aid=plL+iSoBABCEO6bMvUYOYT\=\= at=checkpointfirewall_ad_opsec dvc=1.22.333.444 deviceZoneURI=/All Zones/ArcSight System/Private Address Space Zones/RFC1918: 1.22.333.444-1.22.333.444 dtz=Europe/Istanbul deviceInboundInterface=eth-s3p1c0 _cefVer=0.1 ad.has__accounting=0CEF:0|Check Point|VPN-1 & FireWall-1||drop|drop|Low| eventId=111111111 msg=Address spoofing proto=UDP categorySignificance=/Informational/Warning categoryBehavior=/Access categoryTechnique=/Traffic Anomaly/Network Layer/Spoof categoryDeviceGroup=/Firewall categoryOutcome=/Failure categoryObject=/Host/Application/Service art=1111111111111 cat=SecurityLog act=drop rt=1111111111111 deviceDirection=0 src=1.22.333.444 sourceZoneURI=/All Zones/ArcSight System/Public Address Space Zones/APNIC/1.22.333.444-1.22.333.444 (APNIC) spt=500 dst=1.22.333.444 destinationZoneURI=/All Zones/ArcSight System/Public Address Space Zones/APNIC/1.22.333.444-1.22.333.444 (APNIC) dpt=500 destinationServiceName=ISAKMP cs1= &  cs3=Mg1-NG cs6=fwgtout-34392011tufin cs1Label=rule & Rule Name cs2Label=UFP category cs3Label=Manager cs4Label=Rule UID cs5Label=Total bytes cs6Label=Policy Name cn1Label=Elapsed Time in Seconds cn2Label=icmp_type cn3Label=icmp_code deviceCustomDate1Label=Local Time c6a2Label=Source IPv6 Address c6a3Label=Destination IPv6 Address ahost=1.22.333.444 agt=1.22.333.444 agentZoneURI=/All Zones/ArcSight System/Private Address Space Zones/RFC1918: 1.22.333.444-1.22.333.444 av=1.22.333.4444.0 atz=Europe/Istanbul aid=flL+iYoBABCAO6bMvUEZYZ\=\= at=checkpointfirewall_ad_opsec dvc=1.22.333.444 deviceZoneURI=/All Zones/ArcSight System/Private Address Space Zones/RFC1918: 1.22.333.444-1.22.333.444 dtz=Europe/Istanbul deviceInboundInterface=eth-s4p2c0 _cefVer=0.1 ad.has__accounting=0

[Former Member · Accepted solution]

Hello,

So if you are reading from a socket, you must have some way of:

1) Determining columns

2) Terminating a record (or message or line):

* Is there a delimiter in the data signifying the end of message?

* Is the data a fixed length?

* Do you read the data one character at a time and keep appending to form a message until you hit some kind of end of message marker?

   InputStreamReader (Java Platform SE 7 )

   BufferedReader (Java Platform SE 7 )

   DataInputStream (Java Platform SE 7 )

ESP's TCP Socket Input Adapter reads a byte buffer (of size determined by the "inputBufferSize" parameter. See: Socket CSV Input Adapter Studio Properties ).  I think the closest publicly available example (including source code) of how this is done is the $ESP_HOME/adapters/framework/examples/streaming_input adapter.  This particular example reads from a file rather than a socket but it is fairly close.

It reads a byte stream from an XML file and sends this byte buffer to the next module using the ESP Toolkit Adapter API's:

     utility.sendRowsBuffer(ByteBuffer.wrap(buf, 0, iRead));

See the following example:

     $ESP_HOME/adapters/framework/examples/src/com/sybase/esp/adapter/framework/examplemodules/ExampleStreamingInputTransporter.java

Then the next module in line has to parse XML elements from the byte buffer and forms a record (AepRecord) to send to the next module:

    $ESP_HOME/adapters/framework/examples/src/com/sybase/esp/adapter/framework/examplemodules/ExampleStreamingInputFormatter.java

The last module in line gets the AepRecord and publishes it to the ESP project.  See:

     $ESP_HOME/adapters/framework/examples/src/com/sybase/esp/adapter/framework/examplemodules/ExampleEspInputTransporter.java

In your data I see a pipe symbol (|) delimits the columns.

But how are you determining when one record/message ends and the next begins?  Counting the number of columns?

[Former Member · Accepted solution]

If I can get CEF messages line-by-line, it is, of course, more easier way but I have been requested to get CEF messages without end-of-line char.

I am working on the Input Formatter to get CEF messages. A CEF message start with "CEF" string. So, I try to parse the incoming stream so that a one line data will be the data stay between two "CEF" string.

Thanks for your comments and advices.

Best regards,

Bulut