Feature: TCP Input #6266

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

+		}
+		r.Reset()
+		// TODO(ph), should we keep the byte array instead of allocating a string?
+		c.forwarder.Send(c.createEvent(scanner.Text()))


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

@@ -0,0 +1,72 @@
+package tcp


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

+  #host: "localhost:8080"
+
+  # Character used to split new message
+  #line_delimiter: "\n"


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

@@ -507,6 +507,16 @@ filebeat.prospectors:
  # Maximum size of the message received over UDP
  #max_message_size: 10240

+#------------------------------ Tcp prospector --------------------------------


+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

+		}
+		r.Reset()
+		// TODO(ph), should we keep the byte array instead of allocating a string?
+		c.forwarder.Send(c.createEvent(scanner.Text()))


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

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

filebeat/input/tcp/client.go

+
+// Handle is reading message from the specified TCP socket.
+func (c *Client) Handle() error {
+	defer c.conn.Close()


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

+	return data
+}
+
+// GetRemoveHosts take the IP address of the client and try to resolve the name, if it fails we


s/getRemoveHosts/getRemoteHosts

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

filebeat/input/tcp/client.go

+
+func (c *Client) cacheMetadata() {
+	c.metadata = common.MapStr{
+		"hostnames":  c.getRemoteHosts(),


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

@@ -0,0 +1,72 @@
+package tcp


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

+import socket
+
+
+class Test(BaseTest):


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

+
+// ScanDelimiter return a function to split line using a custom delimiter, the delimiter
+// is stripped from the returned value.
+func ScanDelimiter(delimiter []byte) bufio.SplitFunc {


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

+  #host: "localhost:8080"
+
+  # Character used to split new message
+  #line_delimiter: "\n"


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

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

filebeat/docs/filebeat-options.asciidoc

+
+[float]
+[[max-read-buffer]]
+==== `max_read_buffer`


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

+#- type: tcp
+  #enabled: false
+  # The host and port to receive the new event
+  #host: "localhost:8080"


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

+
+func (c *Client) cacheMetadata() {
+	c.metadata = common.MapStr{
+		"hostnames":  c.getRemoteHosts(),


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

+		Type: "tcp",
+	},
+	LineDelimiter: "\n",
+	Host:          "localhost:9000",


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

+	},
+	LineDelimiter: "\n",
+	Host:          "localhost:9000",
+	Timeout:       5 * 60,


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

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

filebeat/input/tcp/conn.go

+	"github.com/pkg/errors"
+)
+
+// ErrMaxReadBuffer returns when too much byte was read on the io.Reader


too many bytes?

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

filebeat/input/tcp/conn.go

+	byteRead      int64
+}
+
+// NewMeteredReader returns a Meteredread


a MeteredReader

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

filebeat/input/tcp/harvester.go

+func (h *Harvester) Run() error {
+	logp.Info("Started listening for incoming TCP connection on: %s:", h.config.Host)
+	for {
+		conn, err := h.server.Accept()


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

+}
+
+// NewDeadlineReader returns a new deadlineReader
+func NewDeadlineReader(c net.Conn, timeout time.Duration) *deadlineReader {


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

+}
+
+// NewMeteredReader returns a new MeteredReader
+func NewMeteredReader(reader io.Reader, maxReadBuffer int64) *meteredReader {


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-20T20:05:35Z

CHANGELOG.asciidoc

@@ -46,8 +46,8 @@ https://github.com/elastic/beats/compare/v6.0.0-beta2...master[Check the HEAD di
 - Refactor the Stdin prospector to use the input interface {pull}6116[6116]
 - Refactor the usage of prospector to input in the YAML reference and
  the system test. {pull}6121[6121]
-


before @ruflin you comment on it :) #6427

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

filebeat/input/tcp/client.go

+// Client is a remote client.
+type Client struct {
+	conn           net.Conn
+	forwarder      *harvester.Forwarder


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

+	data := util.NewData()
+	data.Event = beat.Event{
+		Timestamp: time.Now(),
+		Meta:      c.metadata,


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

+
+// GetRemoteHosts take the IP address of the client and try to resolve the name, if it fails we
+// fallback to the IP, IP can resolve to multiple hostname.
+func (c *Client) getRemoteHosts() []string {


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

+	return hosts
+}
+
+func (c *Client) cacheMetadata() {


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

+)
+
+// Harvester represent a TCP server
+type Harvester struct {


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

+				conn.RemoteAddr(),
+				h.clientsCount(),
+			)
+			h.wg.Done()


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

ph force-pushed the feature/tcp-input branch from 70507c4 to eee70ef Compare February 2, 2018 15:46

ph force-pushed the feature/tcp-input branch from f1b87ea to 6219e97 Compare February 2, 2018 19:31

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

ph force-pushed the feature/tcp-input branch from 6219e97 to cb1596c Compare February 5, 2018 14:33

ruflin reviewed Feb 6, 2018

View reviewed changes

monicasarbu mentioned this pull request Feb 8, 2018

Beat for tcp input / proxy #536

Closed

ph force-pushed the feature/tcp-input branch from 9d904f1 to 7f9ecdf Compare February 9, 2018 02:21

ph force-pushed the feature/tcp-input branch from 7f9ecdf to 558d422 Compare February 12, 2018 19:14

houndci-bot reviewed Feb 12, 2018

View reviewed changes

ph force-pushed the feature/tcp-input branch 2 times, most recently from d2f4e36 to c0b79db Compare February 12, 2018 19:28

ph force-pushed the feature/tcp-input branch from c0b79db to 8028c8d Compare February 12, 2018 20:44

ph force-pushed the feature/tcp-input branch 2 times, most recently from 338edb7 to 435069e Compare February 13, 2018 17:50

ph commented Feb 20, 2018

View reviewed changes

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

ph force-pushed the feature/tcp-input branch from a0a3349 to 7b920dd Compare February 23, 2018 21:14

dedemorton mentioned this pull request Mar 13, 2018

Refactor input docs #6537

Merged

3 tasks

ph added 10 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

ph force-pushed the feature/tcp-input branch from 7b920dd to fd41e40 Compare March 27, 2018 15:29

Update making Host mandatory.

98d521e

ph force-pushed the feature/tcp-input branch from fd41e40 to 98d521e Compare March 27, 2018 15:42

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

Feature: TCP Input #6266

Feature: TCP Input #6266

Conversation

ph commented Feb 2, 2018 • edited Loading

ph commented Feb 2, 2018

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ph commented Feb 2, 2018

ruflin left a comment • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ph Feb 5, 2018 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ph commented Feb 5, 2018

ph commented Feb 5, 2018

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ph Feb 6, 2018 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ph Feb 6, 2018 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ruflin commented Feb 11, 2018

ph commented Feb 12, 2018

ph commented Feb 12, 2018

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ph commented Feb 12, 2018

ruflin commented Feb 13, 2018

ph commented Feb 13, 2018

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

urso Feb 21, 2018 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

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

urso Feb 21, 2018 •

edited

Loading

ph commented Mar 27, 2018 •

edited

Loading