Why is the data lost in elasticsearch ? #1141
Comments
|
there's a "/metrics" endpoint. It would show how many spans were accepted by kafka on the zipkin-server side. can you send relevant output from that? also, I don't understand what you mean by
do you mean that when you post a span where the id is all bits set, it comes back not in hex? |
|
I set id from 1-500000 in hex. I think that the index have a conflict. |
|
The lost data is different for storing in the elasticsearch every, when I wrote data in the kafka. |
|
wondering if this is to do with the data being older? Ex. if I post this, I have to set lookback to a relatively high value to see it. $ curl -s localhost:9411/api/v1/spans -X POST -H "Content-Type: application/json" --data '[
{
"traceId": "ffffffffffffffff",
"name": "fermentum",
"id": "ffffffffffffffff",
"annotations": [
{
"timestamp": 1466386794757000,
"value": "sr",
"endpoint": {
"serviceName": "semper",
"ipv4": "113.29.89.129",
"port": 2131
}
},
{
"timestamp": 1466386794757000,
"value": "sagittis",
"endpoint": {
"serviceName": "semper",
"ipv4": "113.29.89.129",
"port": 2131
}
},
{
"timestamp": 1466386794758000,
"value": "montes",
"endpoint": {
"serviceName": "semper",
"ipv4": "113.29.89.129",
"port": 2131
}
},
{
"timestamp": 1466386794758000,
"value": "augue",
"endpoint": {
"serviceName": "semper",
"ipv4": "113.29.89.129",
"port": 2131
}
},
{
"timestamp": 1466386794759000,
"value": "malesuada",
"endpoint": {
"serviceName": "semper",
"ipv4": "113.29.89.129",
"port": 2131
}
},
{
"timestamp": 1466386794760000,
"value": "ss",
"endpoint": {
"serviceName": "semper",
"ipv4": "113.29.89.129",
"port": 2131
}
}
],
"binaryAnnotations": [
{
"key": "mollis",
"value": "hendrerit",
"endpoint": {
"serviceName": "semper",
"ipv4": "113.29.89.129",
"port": 2131
}
}
]
}
]'
$ curl -s 'localhost:9411/api/v1/traces?serviceName=semper&lookback=5000000000'|jq . |
|
I started a local server against elasticsearch and got the same output. ex the trace id returned the same as if it were not using elasticsearch $ SELF_TRACING_ENABLED=false STORAGE_TYPE=elasticsearch java -jar zipkin.jar |
|
@liangman we need some way of reproducing the problem, so maybe verify versions and see if you can reproduce something using POST like above. Our tests run latest zipkin against elasticsearch 2.2.1 |
|
Ok, It is normal for using "POST", but why is the data lost for using "kafka + ES"? |
|
great... glad to see progress. First step is to make sure that when ex. http://localhost:9411/metrics https://github.com/openzipkin/zipkin/tree/master/zipkin-server#collector Once you verify this, change only the transport variable and run the look at the /metrics endpoint and compare the kafka stats with the https://github.com/openzipkin/zipkin/tree/master/zipkin-server#logging You might see dropped spans or dropped messages in the metrics output, |
|
Of course, it was the "POST + ES"; |
|
I had sent the same 100 data with kafka and http. http+es: |
|
looking at your metrics output, it seems you aren't running the latest
version of zipkin (counter.zipkin_collector.bytes should have read
counter.zipkin_collector.bytes.http).
I don't think this impacts your issue, but it would be less distracting to
use the same version of code (latest is 1.1.5).
One thing that seems odd is that the cumulative bytes collected from
http(157k) are less than the cumulative bytes collected from kafka (407k).
Are you using the same encoding for both? the byte count is after any
decompression, so I'd expect figures to be similar...
Regardless, if a scenario of only 100 spans can create the concern, it
seems small enough to be something myself or someone else could run with
ease.
do you mind posting your script somewhere so that I can try it?
|
|
senddata.sh: #!/bin/bash
TIMESTAMP=$(node -e 'console.log(new Date().getTime())')000
curl -s localhost:9411/api/v1/spans -X POST -H "Content-Type: application/json" --data '[{
"traceId": "'${1}'",
"name": "fermentum",
"id": "'${1}'",
"annotations": [
{
"timestamp": '${TIMESTAMP}',
"value": "sr",
"endpoint": {
"serviceName": "semper",
"ipv4": "113.29.89.129",
"port": 2131
}
},
{
"timestamp": '${TIMESTAMP}',
"value": "sagittis",
"endpoint": {
"serviceName": "semper",
"ipv4": "113.29.89.129",
"port": 2131
}
},
{
"timestamp": '${TIMESTAMP}',
"value": "montes",
"endpoint": {
"serviceName": "semper",
"ipv4": "113.29.89.129",
"port": 2131
}
},
{
"timestamp": '${TIMESTAMP}',
"value": "augue",
"endpoint": {
"serviceName": "semper",
"ipv4": "113.29.89.129",
"port": 2131
}
},
{
"timestamp": '${TIMESTAMP}',
"value": "malesuada",
"endpoint": {
"serviceName": "semper",
"ipv4": "113.29.89.129",
"port": 2131
}
},
{
"timestamp": '${TIMESTAMP}',
"value": "ss",
"endpoint": {
"serviceName": "semper",
"ipv4": "113.29.89.129",
"port": 2131
}
}
],
"binaryAnnotations": [
{
"key": "mollis",
"value": "hendrerit",
"endpoint": {
"serviceName": "semper",
"ipv4": "113.29.89.129",
"port": 2131
}
}
]
}]' |
|
start_send.sh: #!/bin/bash
i=1
while [ $i -lt $1 ]
do
echo $i
./senddata.sh `printf "%x" $i`
let "i=${i}+1"
done |
|
./start_send.sh 101 |
|
I use the java for writing the data in kafka, so I don't know how i post it. But I can write the script with the python. Please wait a moment. |
|
maybe you can send the same json you send via http using something like $ kafka-console-producer.sh --broker-list $ADVERTISED_HOST:9092 --topic zipkin
[{"traceId":"1","name":"bang","id":"2","timestamp":1234,"binaryAnnotations":[{"key":"lc","value":"bamm-bamm","endpoint":{"serviceName":"flintstones","ipv4":"127.0.0.1"}}]}]https://github.com/openzipkin/zipkin/tree/master/zipkin-collector/kafka#json |
|
senddatatokafka.py: #!/bin/python
import time
import os
data=data="""[{
"traceId": "%x",
"name": "fermentum",
"id": "%x",
"annotations": [
{
"timestamp": %i,
"value": "sr",
"endpoint": {
"serviceName": "semper",
"ipv4": "113.29.89.129",
"port": 2131
}
},
{
"timestamp": %i,
"value": "sagittis",
"endpoint": {
"serviceName": "semper",
"ipv4": "113.29.89.129",
"port": 2131
}
},
{
"timestamp": %i,
"value": "montes",
"endpoint": {
"serviceName": "semper",
"ipv4": "113.29.89.129",
"port": 2131
}
},
{
"timestamp": %i,
"value": "augue",
"endpoint": {
"serviceName": "semper",
"ipv4": "113.29.89.129",
"port": 2131
}
},
{
"timestamp": %i,
"value": "malesuada",
"endpoint": {
"serviceName": "semper",
"ipv4": "113.29.89.129",
"port": 2131
}
},
{
"timestamp": %i,
"value": "ss",
"endpoint": {
"serviceName": "semper",
"ipv4": "113.29.89.129",
"port": 2131
}
}
],
"binaryAnnotations": [
{
"key": "mollis",
"value": "hendrerit",
"endpoint": {
"serviceName": "semper",
"ipv4": "113.29.89.129",
"port": 2131
}
}
]
}]"""
def main():
count = 100
data1=data.replace(' ', '').replace('\n', '')
cmdp = r'./bin/kafka-console-producer.sh --broker-list localhost:9092 --topic zipkin'
pipp = os.popen(cmdp, 'w')
i = 0
while i < count:
i += 1
timestamp = time.time() * 10 ** 6
pipp.write(data1%(i, i, timestamp, timestamp, timestamp, timestamp, timestamp, timestamp) + "\r\n")
#print data1%(i, i, timestamp, timestamp, timestamp, timestamp, timestamp, timestamp)
print 'finsh!'
pipp.close()
if __name__ == '__main__':
main()python senddatatokafka.py |
|
Beause I remove the " " and "\n" for sending to kafka. I think that you can try it with the script. |
|
OK so I've verified that with the following setup, I get readbacks between 58 and 100 spans when using the kafka script vs the http one which routinely reads back 100. what I do, is run the scenarios below while kafka is left up, but elasticsearch is cleaned between runs Where below instructions run kafka and ES # start kafka
$ curl -SL http://www.us.apache.org/dist/kafka/0.8.2.2/kafka_2.11-0.8.2.2.tgz | tar xz
$ nohup bash -c "cd kafka_* && bin/zookeeper-server-start.sh config/zookeeper.properties >/dev/null 2>&1 &"
$ nohup bash -c "cd kafka_* && bin/kafka-server-start.sh config/server.properties >/dev/null 2>&1 &"
# start ES
$ curl -SL https://download.elasticsearch.org/elasticsearch/release/org/elasticsearch/distribution/tar/elasticsearch/2.2.1/elasticsearch-2.2.1.tar.gz | tar xz
$ elasticsearch-*/bin/elasticsearch -d > /dev/nullAnd.. I start zipkin-server like so.. SELF_TRACING_ENABLED=false KAFKA_ZOOKEEPER=localhost:2181 STORAGE_TYPE=elasticsearch java -jar zipkin.jar --logging.level.zipkin=DEBUGHTTPWhen I run the HTTP test like this: for i in {1..100}; do ./senddata.sh `printf "%x" $i`; doneI get these collector metrics: And the api count looks correct: $ curl -s 'localhost:9411/api/v1/traces?lookback=500000000&limit=100'|jq '. | length'
100KafkaWhen I run the Kafka test like this: $ python senddatatokafka.py
Java HotSpot(TM) 64-Bit Server VM warning: Option UseParNewGC was deprecated in version 9.0 and will likely be removed in a future release.
[2016-06-25 14:11:06,504] WARN Property topic is not valid (kafka.utils.VerifiableProperties)
finsh!
[2016-06-25 14:11:06,676] WARN Error while fetching metadata [{TopicMetadata for topic zipkin ->
No partition metadata for topic zipkin due to kafka.common.LeaderNotAvailableException}] for topic [zipkin]: class kafka.common.LeaderNotAvailableException (kafka.producer.BrokerPartitionInfo)
[2016-06-25 14:11:06,684] WARN Error while fetching metadata [{TopicMetadata for topic zipkin ->
No partition metadata for topic zipkin due to kafka.common.LeaderNotAvailableException}] for topic [zipkin]: class kafka.common.LeaderNotAvailableException (kafka.producer.BrokerPartitionInfo)
[2016-06-25 14:11:06,684] ERROR Failed to collate messages by topic, partition due to: Failed to fetch topic metadata for topic: zipkin (kafka.producer.async.DefaultEventHandler)I get these collector metrics: And the api count looks correct sometimes, and not others (always the stats look the same): $ curl -s 'localhost:9411/api/v1/traces?lookback=500000000&limit=100'|jq '. | length'
100 |
|
NEXT STEP: one difference between in-memory storage and ES storage is that the former doesn't do anything asynchronously. We should validate that this scenario against Cassandra, too (as it also uses guava futures). |
|
en, I will try it again following you step. |
|
Might be an issue in cassandra, too, but looks like #1142 is blocking my ability to use the normal readback (only returns 10-16) |
|
No, when I send 100 data to the kafka, all data is written in cassandra. But I view 10 data from the page of zipkin. Maybe there is a bug in the code... |
|
@liangman I edited the comment for the http script. can you edit the one for kafka and make sure that timestamps are reset each time (using epoch micros)? |
|
NEXT STEP: See the behavior when the kafka script reports spans with unique timestamps. For example A step after that would be to instrument ps I'm offline likely the rest of the day, but might look into this tomorrow. |
|
sure.. I'd like you to edit your comment here #1141 (comment) in the span you are generating in python, please make it have timestamps according to current system time. That reduces the work needed when querying and also ttl considerations. For example, you can look at how I edited the http script. after that you can try to troubleshoot ElasticsearchSpanConsumer by customizing the class, building and running locally. https://github.com/openzipkin/zipkin/tree/master/zipkin-server#running-locally If you aren't familiar enough to do that, you'd likely need to wait until I have more time to help (or someone else does). |
|
I have edited the script again. It may meet your requirements. |
|
thanks. will take a look today On Sat, Jun 25, 2016 at 5:19 PM, liangman [email protected] wrote:
|
|
Update: Adding a 100ms sleep between kafka messages seems to avoid this issue.. There's no errors, but something seems to drop when processing concurrently. |
|
Note: synchronizing ElasticsearchSpanConsumer doesn't make the problem go away.. there's something else going on that sleeping 100ms between creating each future avoids. |
|
I've found a workaround. This happens when a bulk request is used for a single request (only 1 span). When I special-case to not use bulk requests when there's only 1 span, the problem disappears. I've two remedies:
We need to create a soak test at some point, too, as the special-casing may not be fixing the root cause, even if it helps. cc @anuraaga |
|
@adriancole i am a bit confused. How come the way messages from kafka are read can affect something specific to ElasticSearch. If i am not wrong, the flow will be kafka -> collector -> ES/C*. From what i read, message loss is only seen for ES and not C*. But the transport (kafka/http) work is done once the data is fetched via collector. So how is the issue related to kafka batching plus ES storage? |
I think http only worked because the test is slower. For example, TL;DR; I would recommend the bundling feature to anyone who seems like Buffering and bundling spans ends up using We've regularly encountered issues with not bundling across different The other option is to see a practice, like dumping many messages at Truth is, we can't count on volunteers to do free support, design and |
During a test where 100 single-span messages are sent to Kafka at the same time, I noticed only 58-97 of them would end up in storage eventhough all messages parsed properly and no operations failed. After a 100ms/message pause was added, the store rate of this test went to 100%, so figured it was some sort of state issue. I noticed the code was using Bulk operations regardless of input size, so as a wild guess changed the special-case single-span messages. At least in this test, it raised the success rate to 100% without any pausing needed. I don't know why this worked, but it seems sensible to not use bulk apis when there's no bulk action to perform. I started to write a unit test to validate single-length lists don't use bulk, but the Mockito involved became too verbose as the Elasticsearch client uses chaining and other patterns that are tedious to mock. Instead, we should make a parallel integration test and apply them to all storage components. See #1141
|
here's the workaround.. when merged it should be testable from snapshot (of course if you are in a position to build the branch, please do |
|
Ok, I have updated the file ElasticsearchSpanConsumer.java. But when I set the count of log-data for sending the kafka, there is an Wranning here: |
|
The result in the ES: |
|
@liangman so let's first make sure we know what does work and what doesn't. Are you saying 100 works, but if you send 250 messages at the same time, you get that rejected exception? since the error includes a capacity of 200, again I'm wondering what would happen if instead of doing 250 spans w/ 250 messages, you instead batched them as 10 or more per message (ex the json array includes 10 items not 1). ps here's the latest snapshot, if it helps http://oss.jfrog.org/oss-snapshot-local/io/zipkin/java/zipkin-server/1.1.6-SNAPSHOT/zipkin-server-1.1.6-20160627.030409-3-exec.jar |
|
Back to the error.. Right now the index queue depth is 200 and you've overrun it. That means requests are being accepted faster than they can be processed. I think it will be useful to adjust the spans/message count to see if you can make storage more efficient with the same topology. There's a lot of reasons it could be backed up, including the usual cpu, mem, network bottlenecks. It could be backed up from a slow cluster even.. We won't be able to troubleshoot what's the bottleneck in your environment, suffice to say you've hit a limit. From an ES point of view, you can look at the tuning options there. Maybe start with this https://www.elastic.co/guide/en/elasticsearch/reference/current/modules-threadpool.html and run experiments until you've found the key parameters that help. |
|
FWIW, it is possible to adjust the default queue size through elasticsearch.yml or cluster settings if you need ...though Adrian beat me to it ;) But as Adrian said, it seems like you are having issues with a highly-synthetic workload, not a real-world one. If this synthetic workload is important to you, then I'd recommend trying to modify the cluster settings, but otherwise would recommend trying to send spans in batch instead (as the Java brave would). And keep in mind, Elasticsearch is IMO a medium-scale datastore. It's not going to get reasonable performance on a single node (this config is only for testing / development) and it's not unlikely you'd run into perf issues with it. |
|
Maybe I have to consider changing the database from ES to cassandra. |
|
When I use the cassandra for storing, it spends about 100s for zipkin consuming 500000 data (per about 1kb). |
|
Thanks for the update.. just curious, was that with a single-node zipkin
against a single-node cassandra cluster?
|
|
I'm closing this issue for now as we've progressed from no-errors and dropped data, to errors that explain why data was dropped (overran queue length), and an ES storage config suggestion to improve that. |
|
ok. |
|
@liangman by the way, thanks your script allowed us to repeat the problem and convert it from a mystery to an explanation. The next users will be better off from your efforts. |
|
I use the latest Zipkin(1.26) and with a single node elasticsearch(2.3.4) for storage(the default docker-zipkin configuration), and still encounter data lost. Here is the test script(sendtozipkin.sh): #!/bin/bash
function send_to_zipkin() {
id=$1
id2=$2
millis=$(python -c 'import time; print "%d" % (time.time() * 1000 * 1000)')
curl localhost:9411/api/v1/spans -X POST -H "Content-Type: application/json" --data '[{
"traceId": "'${id}'",
"name": "fermentum",
"id": "'${id}'",
"annotations": [
{
"timestamp": '${millis}',
"value": "sr",
"endpoint": {
"serviceName": "semper",
"ipv4": "113.29.89.129",
"port": 2131
}
}
]
}, {
"traceId": "'${id}'",
"name": "fermentum1",
"id": "'${id2}'",
"annotations": [
{
"timestamp": '${millis}',
"value": "sr",
"endpoint": {
"serviceName": "semper",
"ipv4": "113.29.89.129",
"port": 2131
}
}
]
}]'
}
i=0
while [ $i -lt $1 ]; do
let "i=${i}+1"
let "j=${i}+16777216"
echo $i
send_to_zipkin `printf "%016x" $i` `printf "%016x" $j`
doneThen send 100(50*2) messages to Zipkin: $ ./sendtozipkin.sh 50And some random messages are lost, I add # grep 'errors\\":true' $(docker inspect --format='{{.LogPath}}' zipkin) | head -1
{"log":"2017-06-12 03:04:29.733 INFO 6 --- [41.133:9200/...] z.s.e.http.ElasticsearchHttpStorage :
{\"took\":1,\"errors\":true,\"items\":[{\"create\":{\"_index\":\"zipkin-2017-06-03\",\"_type\":\"span\",
\"_id\":\"AVyaQoUkPGYSeXOvPsRT\",\"status\":429,\"error\":{\"type\":\"es_rejected_execution_exception\",
\"reason\":\"rejected execution of org.elasticsearch.transport.TransportService$4@5742b77e on EsThreadPoolExecutor[bulk, queue capacity = 50, org.elasticsearch.common.util.concurrent.EsThreadPoolExecutor@2fcaec02[Running, pool size = 8, active threads = 8, queued tasks = 50, completed tasks = 0]]\"}}},
{\"index\":{\"_index\":\"zipkin-2017-06-03\",\"_type\":\"servicespan\",
\"_id\":\"gamerebategift|timeraddpoolamounttask\",\"status\":429,\"error\":{\"type\":\"es_rejected_execution_exception\",
\"reason\":\"rejected execution of org.elasticsearch.transport.TransportService$4@70fc3bfe on EsThreadPoolExecutor[bulk, queue capacity = 50, org.elasticsearch.common.util.concurrent.EsThreadPoolExecutor@2fcaec02[Running, pool size = 8, active threads = 8, queued tasks = 50, completed tasks = 0]]\"}}}]}\n",
"stream":"stdout","time":"2017-06-12T03:04:29.733933728Z"}Check the default configuration: $ curl -XGUT localhost:9200/_cluster/settings
{"persistent":{},"transient":{}}Change $ curl -XPUT localhost:9200/_cluster/settings -d '{"transient": {"threadpool.bulk.queue_size": 500}}'
{"acknowledged":true,"persistent":{},"transient":{"threadpool":{"bulk":{"queue_size":"500"}}}}Then rerun the sendtozipkin.sh script, there were no data lost any more. |
|
What you are saying is that when there's an overload on the elasticsearch
cluster, zipkin drops data. The only recourse besides tuning ES would be to
buffer (or push-back if using kafka), right?
|
|
Yes, at least Zipkin should log a WARNING/ERROR log if writing ES with errors returned while not setting |
|
Yes, and I think Zipkin should log a WARNING/ERROR log if writing ES with
errors returned while not setting ES_HTTP_LOGGING=BODY.
logging during collection, especially at warning or error level can fill
logs and cause even more problems. Not sure this will be a good solution.
Usually, you'd monitor for dropped messages or such by monitoring collector
metrics. logging failures at debug or trace may be fine either way.
Notes on collector metrics:
https://github.com/openzipkin/zipkin/blob/master/zipkin-server/README.md#collector
Example dashboard:
openzipkin-attic/docker-zipkin#135 (comment)
|
I used the zipkin, kafka, and elasticsearch for testing. The elasticsearch was only a node. It wasn't problem that the data was transferred from kafka to zipkin, but the data is lost from zipkin to elasticsearch. I wrote 500000 log-data to kafka, but it was only 212162 in the elasticsearch.
The data:
[{"traceId":"ffffffffffffffff","name":"fermentum","id":"ffffffffffffffff","annotations":[{"timestamp":1466386794757000,"value":"sr","endpoint":{"serviceName":"semper","ipv4":"113.29.89.129","port":2131}},{"timestamp":1466386794757000,"value":"sagittis","endpoint":{"serviceName":"semper","ipv4":"113.29.89.129","port":2131}},{"timestamp":1466386794758000,"value":"montes","endpoint":{"serviceName":"semper","ipv4":"113.29.89.129","port":2131}},{"timestamp":1466386794758000,"value":"augue","endpoint":{"serviceName":"semper","ipv4":"113.29.89.129","port":2131}},{"timestamp":1466386794759000,"value":"malesuada","endpoint":{"serviceName":"semper","ipv4":"113.29.89.129","port":2131}},{"timestamp":1466386794760000,"value":"ss","endpoint":{"serviceName":"semper","ipv4":"113.29.89.129","port":2131}}],"binaryAnnotations":[{"key":"mollis","value":"hendrerit","endpoint":{"serviceName":"semper","ipv4":"113.29.89.129","port":2131}}]}]"ffffffffffffffff" was replaced in "1-5000000";
It was no regular!!!
The text was updated successfully, but these errors were encountered: