Feature: TCP Input by ph · Pull Request #6266 · elastic/beats

ph · 2018-02-02T14:40:04Z

Allow to receive new events via TCP, this will create a new event per
line and add some useful information about the connected client to the
evant. This input is marked as experimental.

This input expose the following settings:

line_delimiter: The characters used to split incoming events, by
default it will split on \n and will also strip both \n and \r.
You can also use any characters like ; and you can also used multiple
characters delimiter like <END>, the delimiter tokens will always be
removed from the string.
max_read_buffer: This is a number of bytes that a client can buffer
in memory before finding a new message, if the limit is reached the
connection is killed and the event is logged. This is to prevent rogue
clients to DoS attack by consuming all the available memory. The default
values is 20971520.
timeout: The server will close any client that reach this inactivity
timeout.

TODO
I've an almost TLS support done, but this PR was getting pretty big and I still have the tests to write so I've excluded it.

Ref: #5862

ph · 2018-02-02T14:40:43Z

Waiting for CI to switch it to review

ph · 2018-02-02T14:41:54Z

filebeat/input/tcp/client.go

This is a legitimate question, I am not sure how the remaining part of the code will work if I keep the message as a byte array.

I would ignore the overhead for now if it adds code complexity.

For me on my side there is no overhead, it just a matter of calling scanner.Bytes() instead of scanner.Text(), I am just unaware of the downstream consequence for processor If I keep the bytes.

Actually, I just read the doc, It's better to do an allocation because the original bytes can be overridden.

// Bytes returns the most recent token generated by a call to Scan. // The underlying array may point to data that will be overwritten // by a subsequent call to Scan. It does no allocation. func (s *Scanner) Bytes() []byte { return s.token }

ph · 2018-02-02T14:42:33Z

filebeat/input/tcp/conn.go

I've keep theses methods public, because I might move it to common.

We should only move it to common if we also use it in metricbeat for example, which could happen.

ph · 2018-02-02T16:36:52Z

Working on fixing the test suite, simple environment difference with how I assert the tests, I will make it more robust. thanks Travis.

ruflin

Only skimmed through it and left some minor comments. Looks really good to me. Happy to dive deeper when it's ready.

BTW: +1 on having TSL in a separate PR.

ruflin · 2018-02-03T03:08:57Z

filebeat/_meta/common.reference.p2.yml

Do we need this to be configurable? I would expect that same behaviour that supports also \r etc. like the log prospector / harvester.

I believe it needs to be configurable; this input is more than just syslog entries.
If you look at the code by default, it will use the ScanLines function to deal with that case. Anything else will use our own implementation of the split.

The default method will detect \n, but it will also strip \r, I think this is the most common use case with plain sockets protocol. So I think we should be fine(tm)?

The custom delimiter is OK for more esoteric implementation. Maybe in a future revision, we could allow multiple patterns.

The feature got requested already several times for the log harvester ...

I haven't look at the log harvester how it does the splitting... maybe we can reuse it.

I did a quick look at the log harvester, I believe we could use the same scanner/split.

ruflin · 2018-02-03T03:10:00Z

filebeat/filebeat.reference.yml

+1 for that change, I've went with Tcp even if golang style would complain because of existing code, I wanted consistency with our Udp input. I guess I can make a PR on top of UDP to be all caps. :D

Yep, saw the UDP one when I reviewed one of your other PR's and was thinking the same.

ruflin · 2018-02-03T03:11:10Z

filebeat/input/tcp/client.go

I would ignore the overhead for now if it adds code complexity.

ruflin · 2018-02-03T03:12:04Z

filebeat/input/tcp/client.go

I wonder if this defer should really be here or outside where the connection is opened or Handle is called.

Make sense, I will that change when you are done with the review.

ruflin · 2018-02-03T03:12:58Z

filebeat/input/tcp/client.go

s/getRemoveHosts/getRemoteHosts

ruflin · 2018-02-03T03:14:27Z

filebeat/input/tcp/client.go

Should we report all hosts? Should we exclude some?

I am not sure what is the best behavior here:

taking the first one could be wrong.

taking a subset could also be wrong.

They are set in the metadata field for now, so users are free to use them or not.

Any chance you could provide some example json output docs in the comment of the PR? I think that makes it easier to understand how the end event will look like.

Also wonder if we want to opt-out from resolving the hostname.

Right it could be costly or slow.

ruflin · 2018-02-03T03:15:19Z

filebeat/input/tcp/conn.go

We should only move it to common if we also use it in metricbeat for example, which could happen.

ruflin · 2018-02-03T03:18:20Z

filebeat/tests/system/test_tcp.py

Nice tests. I already head in mind to ask for this, especially seeing if the shut down works 👍

ruflin · 2018-02-03T03:19:17Z

libbeat/common/scan.go

I would not put functions into common which we only use in one beat. We should have at least 2 use cases.

ph · 2018-02-05T14:34:48Z

@ruflin You can go ahead with the full review, I was just marking it as in progress to wait for CI. Since I was testing some goroutine interaction I wanted to make it was solid on the CI. I haven't made your initial changes to this PR yet. But go a full review and I will make them in batch.

Thanks for checking it out.

ph · 2018-02-05T18:14:23Z

jenkins test this please

ruflin · 2018-02-06T00:12:07Z

filebeat/_meta/common.reference.p2.yml

The feature got requested already several times for the log harvester ...

ruflin · 2018-02-06T00:13:33Z

filebeat/docs/filebeat-options.asciidoc

We have max_message_size in udp. We should probably use the same config name. Also we can change the udp one.

I will rename max_read_buffer to max_message_size, after looking back at the implementation they are a bit different but the underlying concept is the same, making sure we don't OOM on incoming messages

ruflin · 2018-02-06T00:16:19Z

filebeat/filebeat.reference.yml

Did you test to have multiple syslog inputs configured? I assume the following should work:

- type: tcp host: "localhost:8080" - type: tcp host: "localhost:8081"

I haven't tested it, but it should work if every input have their own port.

If both port are free:

2018-02-07T11:47:40.997-0500 INFO tcp/harvester.go:58 Started listening for incoming TCP connection on: localhost:8080` 2018-02-07T11:47:40.997-0500 INFO input/input.go:87 Starting input of type: tcp; ID: 14733937925171989371 2018-02-07T11:47:40.997-0500 DEBUG [processors] processors/processor.go:49 Processors: 2018-02-07T11:47:40.997-0500 INFO tcp/input.go:50 Starting TCP input 2018-02-07T11:47:40.997-0500 INFO tcp/harvester.go:58 Started listening for incoming TCP connection on: localhost:8081 2018-02-07T11:47:40.998-0500 DEBUG [input] log/config.go:178 recursive glob enabled

if one port is already taken:

2018-02-07T11:48:52.928-0500 ERROR instance/beat.go:674 Exiting: Error in initing input: listen tcp 127.0.0.1:8080: bind: address already in use Exiting: Error in initing input: listen tcp 127.0.0.1:8080: bind: address already in use

FB exits

ruflin · 2018-02-06T00:22:23Z

filebeat/input/tcp/client.go

Any chance you could provide some example json output docs in the comment of the PR? I think that makes it easier to understand how the end event will look like.

ruflin · 2018-02-06T00:24:06Z

filebeat/input/tcp/config.go

Is 9000 a "common" default?

@ruflin Not sure about that, I think it's has good as the default we have in UDP. :)

ruflin · 2018-02-06T00:24:49Z

filebeat/input/tcp/config.go

I think you will have to define the unit here (time.Second)

ruflin · 2018-02-06T00:25:15Z

filebeat/input/tcp/conn.go

too many bytes?

ruflin · 2018-02-06T00:26:04Z

filebeat/input/tcp/conn.go

a MeteredReader

ruflin · 2018-02-06T00:30:27Z

filebeat/input/tcp/harvester.go

I would expect that the prospector/input would be the server and then each harvester would be a client?

Like we discussed in the input refactoring, I am not sure the wording harvester work in the context of UDP and TCP, I would support renaming harvester to server and use client for the actual remote connection. Would you be OK with that?

My reasoning is harvester is more active process to fetch new things (pull), which on the other hand a TCP / UDP is more passive (push).

I'm not really concerned here about not using the name harvester but why input != server. From my point of view 1 input = 1 server, so if I start an input that should mean the server is started. Instead of having input, server, client I think we only need input=server, client and can remove one abstraction layer.

I'm not too concerned about not having the name harvester anywhere as this is not something exposed to the user.

@ph Now looking at my own UDP code I see that I used also the harvester as the server :-( Leading by bad example? :-(

Not, you lead by good example you see the problem :)

ruflin · 2018-02-11T23:25:26Z

@ph I suggest we get this PR in rather sooner then later. The internal code structure I brought up above can be discussed and refactored later especially as I realised now the changes would also apply to UDP. Can you go through the other comments above again and see what needs addressing, rebase and squash and ping me for an other review and merge?

ph · 2018-02-12T17:01:00Z

@ruflin Thanks for answering my comments will do the changes, I wanted to bulk them. :)

ph · 2018-02-12T19:13:17Z

This is an example of the event format from a received events with this input.

 {
  "@timestamp": "2018-02-12T19:12:29.000Z",
  "@metadata": {
    "beat": "filebeat",
    "type": "doc",
    "version": "7.0.0-alpha1",
    "hostnames": [
      "localhost"
    ],
    "ip_address": "127.0.0.1:56576"
  },
  "beat": {
    "hostname": "sashimi",
    "version": "7.0.0-alpha1",
    "name": "sashimi"
  },
  "message": "Hello world",
  "input": {
    "type": "tcp"
  },
  "prospector": {
    "type": "tcp"
  }
}

houndci-bot · 2018-02-12T19:14:41Z

filebeat/input/tcp/conn.go

exported func NewDeadlineReader returns unexported type *tcp.deadlineReader, which can be annoying to use

houndci-bot · 2018-02-12T19:14:41Z

filebeat/input/tcp/conn.go

exported func NewMeteredReader returns unexported type *tcp.meteredReader, which can be annoying to use

ph · 2018-02-12T20:15:28Z

I am fixing the flaky tests, oh network services.

ruflin · 2018-02-13T01:07:53Z

Thanks for the json output. We should probably also add an example to the docs. I'm a bit on the fence about the @metadata usage but it's something we can figure out later.

ph · 2018-02-13T17:57:53Z

@ruflin Let's do a following PR for the JSON example in the doc, not sure where it would be the best place to insert it, inputs doesn't dedicated pages yet, Any idea @dedemorton where I could place this information ?

ph · 2018-02-13T18:24:34Z

@ruflin Go ahead with the review, there is a failure in the Filebeat test with x-pack, I think the available snapshot isn't correct right now..

ERROR: This plugin was built with an older plugin structure. Contact the plugin author to remove the intermediate "elasticsearch" directory within the plugin zip.

ruflin

LGTM. All the comments could also be addressed in a follow up PR. I would like to get this out rather soonish so we get in the hands of our users and get feedback for it.

@urso Are you ok with merging this or want to have a look first?

ruflin · 2018-02-14T00:08:47Z

CHANGELOG.asciidoc

Should this line be here?

ruflin · 2018-02-14T00:09:55Z

filebeat/docs/filebeat-options.asciidoc

As we don't have separate pages yet for inputs I wonder if we should mark here that this feature is actually experimental.

@dedemorton Thoughts?

ruflin · 2018-02-14T00:10:49Z

filebeat/docs/filebeat-options.asciidoc

Good for now but we now have really the need to split up config options per input (@dedemorton I remember you were thinking about this).

@ruflin Looking at this is at the top of my list for next week.

ruflin · 2018-02-14T00:11:05Z

filebeat/docs/filebeat-options.asciidoc

We should mention that this is for tcp only.

ruflin · 2018-02-14T00:11:49Z

filebeat/input/tcp/client.go

Could be a unit64?

ruflin · 2018-02-14T00:21:22Z

filebeat/input/tcp/harvester.go

This should be logp.Debug("tcp", "Can not accept the connection: %s", err) I think

ruflin · 2018-02-14T00:22:10Z

filebeat/input/tcp/harvester.go

Seeing that you also close the connection here, I wonder what happens if Close is called twice?

ruflin · 2018-02-14T00:23:40Z

filebeat/input/tcp/harvester.go

Should each client have a unique id instead? Could make it interesting for stats later on.

I think it still fine in that case, I don't think a unique ID like a UUID will give us more than using the ptr.

ruflin · 2018-02-14T00:25:54Z

filebeat/input/tcp/input.go

Want to add the host info here?

ruflin · 2018-02-14T00:26:06Z

filebeat/input/tcp/input.go

Add the host info?

kvch · 2018-02-14T16:09:33Z

filebeat/input/tcp/config.go

Why don't you write 5 * time.Minute instead of calculating it?

I can and will.

kvch · 2018-02-14T16:14:25Z

filebeat/input/tcp/conn.go

I am not a big fan of this name. It would be nice to express that it is a special LimitedReader.
However, I am not able to come up with a proper name. My best alternative proposal is ResetterLimitedReader. I am willing to let the change go.

Naming is hard but, ResetterLimitedReader or ResetableLimitedReader. I am ok with both.

kvch · 2018-02-15T10:07:56Z

filebeat/input/tcp/harvester.go

I think you could eliminate calling allClients and simply loop over the map:

for client, _ := range h.clients { client.Close() }

At this point there can't be any new connection, right? Does locking make sense here? If locking is required, I am ok with keeping the function. But the transformation seems unnecessary.

When this code is run we stop refusing new connection but, when the client is shutting down each client are removed from the hash at the end of the goroutine.

So I have two options either I use a copy of the original hash which is what allClients is doing or I lock before looping the element in the array this will make all the goroutine wait for the mutex before they can remove themself. I went with the former to keep consistency in the values in the hash during the loop.

kvch · 2018-02-15T10:09:46Z

filebeat/input/tcp/harvester_test.go

+1 for extracting this function

I will wait when we add a bit more feature to it.

kvch · 2018-02-15T10:14:10Z

filebeat/input/tcp/harvester_test.go

Nitpicking, I am completely fine if you ignore this comment:
I have seen that you used top-down design, I am not sure if it was on purpose, it made reviewing your code much easier. Do you mind reordering the functions here, so it would follow the same pattern?

ph · 2018-02-20T19:04:22Z

@ruflin @kvch I have update this PR with your comments, tests were running locally but I will keep an eye on the CI.

ph · 2018-02-20T20:05:35Z

CHANGELOG.asciidoc

before @ruflin you comment on it :) #6427

urso · 2018-02-21T14:30:48Z

filebeat/input/tcp/client.go

can we replace *harvester.Forwarder by an interface or function type? Some more isolation from harvester.Forwarder (I hope to remove Forwarder and Outlet in the future).

Knowing that it make sense to change it.

Similar to #6439 ?:)
I was planning to do it in second part but I can do it right now.

let's keep the PR as small as possible and do the interface when we have a second implementation.

urso · 2018-02-21T14:38:23Z

filebeat/input/tcp/client.go

The metadata field behaves like @metadata in Logstash. If one sets index or id or pipeline, this will be configure some paramters in the Elasticsearch output. This on purpose?

When indexing into ES, the @metadata is dropped. That is, when sending directly to ES, this information can never be retrieved by users (no rename processor).

I did it on purpose, I am not sure how valuable that information it is. but I could make it public field and store in the mapping. WDYT?

urso · 2018-02-21T14:41:10Z

filebeat/input/tcp/client.go

do we need a method on Client for this one?

I don't see the need to make it a simple function or if It can be reused elsewhere, it will called once and we cache the result.

urso · 2018-02-21T14:42:43Z

filebeat/input/tcp/client.go

cacheMetadata is quite short and only executed once by the constructor. How about moving the body into the constructor and removing the function.

Alternatively have it be a function returning a common.MapStr?

I will go with the common.MapStr.

urso · 2018-02-21T14:53:27Z

filebeat/input/tcp/harvester.go

Let us not export the Harvester. It's some internal detail specific to the input. How about renaming to server?

Do we actually need a special Harvester/server type? The Input implementations seems to mostly wrap the server.

I would leave it as is and rename the harvester to server similar when doing #6361 similar to #6439, But I can do it right now if you wish.

urso · 2018-02-21T14:57:06Z

filebeat/input/tcp/harvester.go

better put defer h.wg.Done() at the top of the go-func. Also try to capture panics and print an error. If something goes really wrong with one client, the other clients can still be served.

Same for h.unregisterClient().

e.g.

go func() { defer logp.Recover() defer h.wg.Done() defer conn.Close() h.registerClient() defer h.unregisterClient() err := client.Handle() ... }()

ruflin · 2018-02-23T07:12:31Z

@ph Can you ping us when this needs an other look?

Allow to receive new events via TCP, this will create a new event per line and add some useful information about the connected client to the evant. This input is marked as **experimental**. This input expose the following settings: - `line_delimiter`: The characters used to split incoming events, by default it will split on `\n` and will also strip both `\n` and `\r`. You can also use any characters like `;` and you can also used multiple characters delimiter like `<END>`, the delimiter tokens will always be removed from the string. - `max_message_size`: This is a number of bytes that a client can buffer in memory before finding a new message, if the limit is reached the connection is killed and the event is logged. This is to prevent rogue clients to DoS attack by consuming all the available memory. The default values is 20971520. - `timeout`: The server will close any client that reach this inactivity timeout. Ref: #5862

ph · 2018-03-27T15:48:44Z

@ruflin @urso

I have rebased this PR and included your latest comments. I also migrated the documentation into his own file.

Open questions:

Should we extract the TCP logic into his own class in this PR, like I did in refactor of #6439?
I would prefer to do it into his own PR and get this one merged.

What do we do about metadata concerning the remote hosts, keep it there, use an actual field from ECS?

I would like to move this forward or at least have a plan to unblock #6433.

ph · 2018-03-27T22:29:24Z

@ruflin @urso Don't check it out yet, I will make some changes I've learned from @andrewkroh

ph · 2018-03-29T13:22:52Z

I think it make sense to reopen a new one, since structure has changes locally to follow the udp refactor.

ph added in progress Pull request is currently in progress. Filebeat Filebeat labels Feb 2, 2018

ph commented Feb 2, 2018

View reviewed changes

ruflin reviewed Feb 3, 2018

View reviewed changes

ph added review and removed in progress Pull request is currently in progress. labels Feb 5, 2018

ruflin reviewed Feb 6, 2018

View reviewed changes

monicasarbu mentioned this pull request Feb 8, 2018

Beat for tcp input / proxy #536

Closed

houndci-bot reviewed Feb 12, 2018

View reviewed changes

ruflin approved these changes Feb 14, 2018

View reviewed changes

kvch reviewed Feb 14, 2018

View reviewed changes

kvch reviewed Feb 15, 2018

View reviewed changes

dedemorton mentioned this pull request Feb 17, 2018

Beats doc updates in 6.x #5632

Closed

37 tasks

ph commented Feb 20, 2018

View reviewed changes

CHANGELOG.asciidoc Outdated

Copy link
Copy Markdown

Contributor Author

ph Feb 20, 2018

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

before @ruflin you comment on it :) #6427

urso reviewed Feb 21, 2018

View reviewed changes

ph mentioned this pull request Feb 22, 2018

[WIP] Add a Syslog input listening to UDP and TCP connection #6433

Closed

dedemorton mentioned this pull request Mar 13, 2018

Refactor input docs #6537

Merged

3 tasks

ph added 11 commits March 27, 2018 10:59

review comments

34acbd3

review from @ruflin

70041bb

config

91ec711

reviews

3882f6d

reorder

8624bdf

use defer instead of direct calls

4c25ecd

move cache metadata in the main body

15d472e

fix

afcf291

update doc with the new format

d9a4ab5

Update making Host mandatory.

98d521e

ph closed this Mar 29, 2018

This was referenced Mar 29, 2018

Feature: TCP Input #6700

Merged

Support a delimiter parameter for the log prospector to allow dividing events on something other than a newline character? #6751

Closed

Conversation

ph commented Feb 2, 2018 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

ph commented Feb 2, 2018

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

ph commented Feb 2, 2018

Uh oh!

ruflin left a comment • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

ph Feb 5, 2018 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

ph commented Feb 5, 2018

Uh oh!

ph commented Feb 5, 2018

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

ph Feb 6, 2018 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

ph commented Feb 2, 2018 •

edited

Loading

ruflin left a comment •

edited

Loading

ph Feb 5, 2018 •

edited

Loading

ph Feb 6, 2018 •

edited

Loading

ph Feb 6, 2018 •

edited

Loading