{"action":"created","issue":{"url":"https://api.github.com/repos/elastic/apm-server/issues/109","repository_url":"https://api.github.com/repos/elastic/apm-server","labels_url":"https://api.github.com/repos/elastic/apm-server/issues/109/labels{/name}","comments_url":"https://api.github.com/repos/elastic/apm-server/issues/109/comments","events_url":"https://api.github.com/repos/elastic/apm-server/issues/109/events","html_url":"https://github.com/elastic/apm-server/pull/109","id":253381082,"number":109,"title":"Respond with 503 immediately if internal queue is full.","user":{"login":"roncohen","id":744,"avatar_url":"https://avatars0.githubusercontent.com/u/744?v=4","gravatar_id":"","url":"https://api.github.com/users/roncohen","html_url":"https://github.com/roncohen","followers_url":"https://api.github.com/users/roncohen/followers","following_url":"https://api.github.com/users/roncohen/following{/other_user}","gists_url":"https://api.github.com/users/roncohen/gists{/gist_id}","starred_url":"https://api.github.com/users/roncohen/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/roncohen/subscriptions","organizations_url":"https://api.github.com/users/roncohen/orgs","repos_url":"https://api.github.com/users/roncohen/repos","events_url":"https://api.github.com/users/roncohen/events{/privacy}","received_events_url":"https://api.github.com/users/roncohen/received_events","type":"User","site_admin":false},"labels":[{"id":661399796,"url":"https://api.github.com/repos/elastic/apm-server/labels/discuss","name":"discuss","color":"fbca04","default":false},{"id":661399781,"url":"https://api.github.com/repos/elastic/apm-server/labels/in%20progress","name":"in progress","color":"d93f0b","default":false}],"state":"open","locked":false,"assignee":null,"assignees":[],"milestone":null,"comments":1,"created_at":"2017-08-28T16:19:02Z","updated_at":"2017-08-28T20:47:28Z","closed_at":null,"pull_request":{"url":"https://api.github.com/repos/elastic/apm-server/pulls/109","html_url":"https://github.com/elastic/apm-server/pull/109","diff_url":"https://github.com/elastic/apm-server/pull/109.diff","patch_url":"https://github.com/elastic/apm-server/pull/109.patch"},"body":"This changes the APM Server to immediately return 503 when the queue is full. I tried this as an experiment to get the desired semantics while requiring no changes to libbeat.\r\n\r\nStart the server:\r\n```\r\n$ ./apm-server -e -E queue.mem.events=40\r\n```\r\n\r\nSend data:\r\n```\r\n$ for i in {0..60}; do curl localhost:8200/v1/errors -w \"$i - %{http_code}\\n\" -d @tests/data/valid/error/minimal_payload_log.json  -H \"Content-Type: application/json\" ; done\r\n0 - 202\r\n1 - 202\r\n2 - 202\r\n3 - 202\r\n4 - 202\r\n5 - 202\r\n6 - 202\r\n7 - 202\r\n8 - 202\r\n9 - 202\r\n10 - 202\r\n11 - 202\r\n12 - 202\r\n13 - 202\r\n...\r\n58 - 202\r\n59 - 202\r\n{\"error\":\"Error adding data to internal queue: Queue is full\"}60 - 503\r\n```\r\n\r\nDetails:\r\n* You'll see it accepts 60 instead 40 events. Long explanation:\r\nI'm reusing `Client` for between requests. The publish callback is protected by a mutex so there's no concurrent usage of the `Client`. I had reuse `Client` because:\r\n  * The code relies on the `DropIfFull` publishing mode which will return immediately if the queue is full: https://github.com/elastic/beats/blob/master/libbeat/publisher/queue/memqueue/produce.go#L121. Each new `Client` has a `Broker` and each `Broker` creates a golang channel with capacity of 20 events: https://github.com/elastic/beats/blob/master/libbeat/publisher/queue/memqueue/broker.go#L85. This channel acts a small buffer that feeds the main queue. If we reuse the `Client`, they share this channel and when it fills up we'll start throwing 503s, which is the desired behaviour. If we don't reuse `Client`, every incoming request will have its own small buffer with capacity for 20 events. \r\n* There's a danger of partially publishing a set of output documents. \r\nSome context: Each event coming into our intake-API can generate multiple outgoing documents. Let's call the output documents resulting from a single incoming event for a \"batch\".  In general, partially dropping a batch is bad because that could lead to inconsistencies in the UI. A small number of short-lived inconsistencies cannot be avoided because of different inherent Elasticsearch/libbeat internals, but if we have a scenario where Elasticsearch has an ingestion rate which is fluctuating around the ingestion rate of the apm-server and the queue has filled up, we can have a problem where batches get half dropped, resulting in significant, permanent inconsistencies in the UI. A better semantic here would be to drop the whole batch if the queue does not have capacity for the all the documents in the batch.\r\n* I don't think any of the above are blockers for moving forward with this."},"comment":{"url":"https://api.github.com/repos/elastic/apm-server/issues/comments/325476658","html_url":"https://github.com/elastic/apm-server/pull/109#issuecomment-325476658","issue_url":"https://api.github.com/repos/elastic/apm-server/issues/109","id":325476658,"user":{"login":"urso","id":75886,"avatar_url":"https://avatars1.githubusercontent.com/u/75886?v=4","gravatar_id":"","url":"https://api.github.com/users/urso","html_url":"https://github.com/urso","followers_url":"https://api.github.com/users/urso/followers","following_url":"https://api.github.com/users/urso/following{/other_user}","gists_url":"https://api.github.com/users/urso/gists{/gist_id}","starred_url":"https://api.github.com/users/urso/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/urso/subscriptions","organizations_url":"https://api.github.com/users/urso/orgs","repos_url":"https://api.github.com/users/urso/repos","events_url":"https://api.github.com/users/urso/events{/privacy}","received_events_url":"https://api.github.com/users/urso/received_events","type":"User","site_admin":false},"created_at":"2017-08-28T20:47:28Z","updated_at":"2017-08-28T20:47:28Z","body":"> There's a danger of partially publishing a set of output documents.\r\nSome context: Each event coming into our intake-API can generate multiple outgoing documents. Let's call the output documents resulting from a single incoming event for a \"batch\". In general, partially dropping a batch is bad because that could lead to inconsistencies in the UI. A small number of short-lived inconsistencies cannot be avoided because of different inherent Elasticsearch/libbeat internals, but if we have a scenario where Elasticsearch has an ingestion rate which is fluctuating around the ingestion rate of the apm-server and the queue has filled up, we can have a problem where batches get half dropped, resulting in significant, permanent inconsistencies in the UI. A better semantic here would be to drop the whole batch if the queue does not have capacity for the all the documents in the batch.\r\n\r\nlibbeat has no idea about the input batch. Even PublishAll forwards one event after another. Single events from multiple clients are potentially merged into one batch. Plus bulk requests to Elasticsearch are limited in the number of documents to be indexed. That is, the problem potentially exists on many levels. For UI consistency reasons I guess we want to publish either all or nothing. \r\n\r\nSome brainstorming following here:\r\n\r\nOne workaround I can think of is introducing another publisher layer with N beat.Client workers + a shared work queue/channel. The workers will use GuaranteedSend for retry and the channel for the workers drops if the queue is not served. e.g.\r\n\r\n```\r\nchDocs := make(chan []beat.Event, 50)\r\nfor i := 0; i < N; i++ {\r\n  go func() { // worker (create N times)\r\n    client, err := beat.Pipeline.ConnectWith(beat.ClientConfig{ PublishMode: beat.GuaranteedMode })\r\n    for docs := range chDocs {\r\n      client.PublishAll(docs) // blocks once broker is full -> no worker available to serve chDocs\r\n    }\r\n  }()\r\n}\r\n\r\n...\r\n\r\ncallback := func(events []beat.Event) error {\r\n  select {\r\n  case chDocs <- events:\r\n    return nil\r\n  default:\r\n    fmt.Errorf(\"channel full\")\r\n  }\r\n}\r\n\r\n```\r\n\r\nYou don't want to use an unbuffered channel, as the select goes with the default case if no worker is already waiting for a client to publish requests (chance to drop events, besides queue is not full).\r\n\r\nDue to creating batches of bulk_max_size, potentially multiple clients' events getting intermixed (if multiple workers are active) and potential out-of-order publishing if load-balancing is enabled, there is still a chance some partial events being published yet. But thanks to infinite retry the missing events will follow (given the beat with in memory queue is not stopped early).\r\n\r\nUsing the ACK handler + storing some intermediate context, one might consider to publish an 'OK' event after all docs have been indexed. Presence of an 'OK' event signals the UI all events are present, plus might contain some more meta-data/metrics (e.g. duration until all events are indexed)."}}
{"actor":{"display_login":"urso"}}