{"action":"created","issue":{"url":"https://api.github.com/repos/elastic/logstash/issues/9225","repository_url":"https://api.github.com/repos/elastic/logstash","labels_url":"https://api.github.com/repos/elastic/logstash/issues/9225/labels{/name}","comments_url":"https://api.github.com/repos/elastic/logstash/issues/9225/comments","events_url":"https://api.github.com/repos/elastic/logstash/issues/9225/events","html_url":"https://github.com/elastic/logstash/pull/9225","id":304213991,"number":9225,"title":"[WIP] Internal Pipelines","user":{"login":"andrewvc","id":131427,"avatar_url":"https://avatars2.githubusercontent.com/u/131427?v=4","gravatar_id":"","url":"https://api.github.com/users/andrewvc","html_url":"https://github.com/andrewvc","followers_url":"https://api.github.com/users/andrewvc/followers","following_url":"https://api.github.com/users/andrewvc/following{/other_user}","gists_url":"https://api.github.com/users/andrewvc/gists{/gist_id}","starred_url":"https://api.github.com/users/andrewvc/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/andrewvc/subscriptions","organizations_url":"https://api.github.com/users/andrewvc/orgs","repos_url":"https://api.github.com/users/andrewvc/repos","events_url":"https://api.github.com/users/andrewvc/events{/privacy}","received_events_url":"https://api.github.com/users/andrewvc/received_events","type":"User","site_admin":false},"labels":[{"id":90997545,"url":"https://api.github.com/repos/elastic/logstash/labels/enhancement","name":"enhancement","color":"009800","default":true},{"id":122184907,"url":"https://api.github.com/repos/elastic/logstash/labels/work%20in%20progress","name":"work in progress","color":"bfe5bf","default":false}],"state":"open","locked":false,"assignee":null,"assignees":[],"milestone":null,"comments":1,"created_at":"2018-03-12T01:15:50Z","updated_at":"2018-03-16T20:10:14Z","closed_at":null,"author_association":"CONTRIBUTOR","pull_request":{"url":"https://api.github.com/repos/elastic/logstash/pulls/9225","html_url":"https://github.com/elastic/logstash/pull/9225","diff_url":"https://github.com/elastic/logstash/pull/9225.diff","patch_url":"https://github.com/elastic/logstash/pull/9225.patch"},"body":"Takes over from https://github.com/logstash-plugins/logstash-integration-internal/pull/1 and is a superset of the changes in #9206 .\r\n\r\nThis WIP PR is the full implementation of inter-pipeline communications in rough form.\r\n\r\nThe main goal here is to let users address complex problems where isolation and encapsulation are desired. Currently, Logstash pipelines are often times complected logical tasks leading to confusing config files. There are a number of cases where this feature is useful that are described below:\r\n\r\n## Use Cases / Patterns\r\n\r\n### The Distributor\r\n\r\nIt is often the case that users would like to use a single input listening on a single port, say, the Beats input, or the HTTP input, to handle different types of data in different ways.\r\n\r\nUsers will often wind up with a pipeline that looks like:\r\n\r\n```\r\ninput { \r\n  beats {}\r\n}\r\n\r\nfilter {\r\n  if ([type] == \"weblogs\") {\r\n    # do weblog stuff\r\n  } else {\r\n     # do generic stuff\r\n  }\r\n}\r\n\r\noutput {\r\n  if ([type] == \"weblogs\") {\r\n    elasticsearch {\r\n      hosts => [\"weblogs.local\"]\r\n    }\r\n  } else {\r\n    elasticsearch {\r\n      hosts => [\"generic.local\"]\r\n    }\r\n  }\r\n}\r\n```\r\n\r\nNote the constant repetition of conditionals. With larger configs, in the thousands of lines, this is quite confusing and hard to reason about. With this patch, we can simplify things considerably by breaking things up into distinct pipelines\r\n\r\n```\r\n# intake.conf\r\ninput {\r\n  beats {}\r\n}\r\n\r\noutput {\r\n  if ([type] == \"weblogs\") {\r\n    internal { send_to => weblogs }\r\n  } else {\r\n    internal { send_to generic_logs }\r\n  }\r\n}\r\n\r\n# weblogs.conf\r\ninput {\r\n  internal { address => weblogs }\r\n}\r\n\r\noutput {\r\n  elasticsearch { hosts => weblogs.local }\r\n}\r\n\r\n\r\n# generic.conf\r\ninput {\r\n  internal { address => generic_logs }\r\n}\r\n\r\noutput {\r\n  elasticsearch { hosts => generic.local }\r\n}\r\n```\r\n\r\n### The Double Buffered Output\r\n\r\nSometimes users have one output, among many, output that has temporary outages. The Logstash pipeline strives not to lose data when an output is blocked, so it blocks and exerts backpressure. If you want to not have this behavior it's not possible today. \r\n\r\nIn the wild: \r\n* [Discuss Thread](https://discuss.elastic.co/t/pipeline-blocking-from-output-plugins/713) \r\n* https://github.com/elastic/logstash/issues/8524  \r\n* [Old Jira-2267](https://logstash.jira.com/browse/LOGSTASH-2267)\r\n\r\n  With internal pipelines it's easy to add buffering to outputs, so that when one goes down the whole pipeline doesn't block. This solves the case where there is a temporary outage.  In the future we could solve allow users to start dropping messages if the PQ is full.\r\n\r\n```\r\n- pipeline.id: senderone\r\n  config.string: \"input { http {}  } filter { #lotsoflogic  } output { internal { send_to => [es, s3] } }\"\r\n  queue.type: persisted\r\n- pipeline.id: sendertwo\r\n  config.string: \"input  { internal { address => es } } output { elasticsearch {} }\"\r\n  queue.type: persisted\r\n- pipeline.id: out\r\n  config.string: \"input { internal { address => s3 } } output { s3 {}  }\"\r\n  queue.type: persisted\r\n```\r\n\r\n\r\n### The Forked Output\r\n\r\nAnother common pattern is used to deal with the situation where one has multiple outputs that should receive different versions of the data.  \r\n\r\nIn the wild: https://discuss.elastic.co/t/can-i-filter-the-output-data-separately/36064/5\r\n\r\nToday these pipelines look like:\r\n \r\n```\r\ninput {\r\n  http { ... }\r\n}\r\n\r\nfilter {\r\n  clone { clones => [\"toS3\"] }\r\n  if [type] == \"toS3\" {\r\n    mutate { add_field => {\"forS3only\" => true}}\r\n  } else {\r\n    mutate { add_field => {\"forESonly\" => true}}\r\n  }\r\n}\r\n\r\noutput {\r\n  if [type] == \"toS3\" {\r\n    s3 {}\r\n  } else {\r\n    elasticsearch {}\r\n  }\r\n}\r\n\r\n```\r\n\r\nWith this patch this can be improved to:\r\n\r\n```\r\n# distributor.conf\r\ninput {\r\n  http { ... }\r\n}\r\n\r\noutput {\r\n  internal { send_to => [s3, es] }\r\n}\r\n\r\n# s3.conf\r\n\r\ninput {\r\n  internal { address => s3 }\r\n}\r\n\r\nfilter {\r\n  mutate { add_field => {\"forS3only\" => true}}\r\n}\r\n\r\noutput {\r\n  s3 { ... }\r\n}\r\n\r\n# es.conf\r\n\r\n\r\ninput {\r\n  internal { address => es }\r\n}\r\n\r\nfilter {\r\n  mutate { add_field => {\"forEsOnly\" => true}}\r\n}\r\n\r\noutput {\r\n  elasticesearch { ... }\r\n}\r\n```\r\n\r\n## Implementation\r\n\r\n It includes three distinct changes:\r\n\r\n* Introduction of the `internal` input / output, which can be used to send messages across pipelines\r\n* Parallel lifecycle events for `Pipeline` lifecycle events, which is necessary to prevent deadlocks (This code is identical to #9206)\r\n* Introduction of Copy on Write (CoW) semantics for `Event#clone`, since that is any time an `Event` crosses a pipeline boundary. Since changes to events must be isolated across pipelines, cheap copies are desirable.\r\n\r\n### The Challenge of CoW semantics\r\n\r\nThe CoW implementation here is only scoped to top-level keys in event data and metadata. If you change a sub-key in the Event, say `[foo][bar]` then the entirety of `[foo]` is copied. Since there is an additional overhead to preserving the CoW structure with an additional `HashSet` tracking which keys are CoW.\r\n\r\nThe CoW implementation in this PR is somewhat broken, though it does meet our API specs. According to our [Event API](https://www.elastic.co/guide/en/logstash/6.2/event-api.html) docs, `Event#get` returns a read-only copy of the event. However, the below pattern is common, even in Logstash core:\r\n\r\n```ruby\r\narrVal = event.get(\"tags\")\r\narrVal << \"foo\"\r\nevent.set(arrVal)\r\n```\r\n\r\nThe returned value is not a read-only copy at all, neither is it a clone of the underlying data. That means that with our 'proper' CoW semantics, as implemented in this patch, the following breaks the CoW promise.\r\n\r\n```ruby\r\noriginal = Event.new\r\ncopy = original.clone\r\narrVal = copy.get(\"tags\")\r\narrVal << \"foo\"\r\ncopy.set(arrVal)\r\noriginal.get(\"tags\") # [\"foo\"]\r\ncopy.get(\"tags\") # [\"foo\"]\r\n```\r\n\r\nThe only practical workaround I can think for this is to make the cloned events CoWoR (copy on write or read), and have `Event#get` return a clone of the specified data. This really negates a lot of the benefits of CoW. We could have the compiled pipeline conditionals use a different `Event#get` that didn't make unnecessary copies, as well as things like `add_tag` that we control. Ultimately, the real problem here is that we return the original mutable datastructures. We could traverse the structures and return an unmodifiable variant but that adds overhead."},"comment":{"url":"https://api.github.com/repos/elastic/logstash/issues/comments/373831056","html_url":"https://github.com/elastic/logstash/pull/9225#issuecomment-373831056","issue_url":"https://api.github.com/repos/elastic/logstash/issues/9225","id":373831056,"user":{"login":"yaauie","id":210924,"avatar_url":"https://avatars1.githubusercontent.com/u/210924?v=4","gravatar_id":"","url":"https://api.github.com/users/yaauie","html_url":"https://github.com/yaauie","followers_url":"https://api.github.com/users/yaauie/followers","following_url":"https://api.github.com/users/yaauie/following{/other_user}","gists_url":"https://api.github.com/users/yaauie/gists{/gist_id}","starred_url":"https://api.github.com/users/yaauie/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/yaauie/subscriptions","organizations_url":"https://api.github.com/users/yaauie/orgs","repos_url":"https://api.github.com/users/yaauie/repos","events_url":"https://api.github.com/users/yaauie/events{/privacy}","received_events_url":"https://api.github.com/users/yaauie/received_events","type":"User","site_admin":false},"created_at":"2018-03-16T20:10:14Z","updated_at":"2018-03-16T20:10:14Z","author_association":"MEMBER","body":"I like the idea behind this a LOT.  :tada:\r\n\r\n - The plugin name `internal` doesn't really ring true to me; is there a reason we don't call this `pipeline`?\r\n\r\n~~~\r\ninput {\r\n  pipeline {\r\n    address => \"foo\"\r\n  }\r\n}\r\n~~~\r\n\r\n~~~\r\noutput {\r\n  pipeline {\r\n    send_to => [\"foo\", \"bar\"]\r\n  }\r\n}\r\n~~~\r\n\r\nI do wonder if we could somehow implement true-CoW using shallow wrappers of [ImmutableRuby](https://github.com/immutable-ruby/immutable-ruby) containers that \"quacked like\" their mutable counterparts. It'd be very very hard to catch all possible edge cases since ruby is ruby, especially in containers-of-containers scenarios, but maybe we could just coerce to those immutable collections for 7.0?\r\n\r\n~~~ ruby\r\nclass COW::Hash < Delegator\r\n  def initialize(hash)\r\n    hash = Immutable::Hash.new(hash) unless hash.kind_of?(Immutable::Hash)\r\n    @reference = hash\r\n    # ...\r\n  end\r\n\r\n  def []=(key, value)\r\n    sync do\r\n      value = @reference.put(key, value)\r\n      @reference = value unless value.__id__ == @reference.__id__\r\n    end\r\n  end\r\n\r\n  # ...\r\nend\r\n~~~\r\n"}}
{"actor":{"display_login":"yaauie"}}