Cygnus ckan-sink throw a error saving the objects

[amfergom]

Hello

I have deployed a Cygnus-ckan instance.

When I update the entity attribute and Orion executes the subscriptions. Cygnus throws the following error:

time=2020-07-16T14:32:37.778Z | lvl=ERROR | corr=2e1f830a-c771-11ea-91ec-e631a1c67994 | trans=de995e73-8d51-41a0-aaa1-f80eefa76a8f | srv=someservice | subsrv=/somesubservicepath | comp=cygnus-ngsi | op=processNewBatches | msg=com.telefonica.iot.cygnus.sinks.NGSISink[579] : CygnusPersistenceError. Could not insert (resId=46bcc158-ca18-49c9-b999-19aa30b46452, statusCode=409). Stack trace: [com.telefonica.iot.cygnus.backends.ckan.CKANBackendImpl.insert(CKANBackendImpl.java:185), com.telefonica.iot.cygnus.backends.ckan.CKANBackendImpl.persist(CKANBackendImpl.java:90), com.telefonica.iot.cygnus.sinks.NGSICKANSink.persistAggregation(NGSICKANSink.java:361), com.telefonica.iot.cygnus.sinks.NGSICKANSink.persistBatch(NGSICKANSink.java:274), com.telefonica.iot.cygnus.sinks.NGSISink.processNewBatches(NGSISink.java:569), com.telefonica.iot.cygnus.sinks.NGSISink.process(NGSISink.java:374), org.apache.flume.sink.DefaultSinkProcessor.process(DefaultSinkProcessor.java:67), org.apache.flume.SinkRunner$PollingRunner.run(SinkRunner.java:145), java.lang.Thread.run(Thread.java:748)]
time=2020-07-16T14:32:42.938Z | lvl=ERROR | corr=12656044-c771-11ea-89ee-e631a1c67994 | trans=N/A | srv=N/A | subsrv=N/A | comp=cygnus-ngsi | op=processRollbackedBatches | msg=com.telefonica.iot.cygnus.sinks.NGSISink[403] : CygnusPersistenceError. Could not insert (resId=8e1af925-76dc-4239-ac32-87b5a8a05e80, statusCode=409). Stack trace: [com.telefonica.iot.cygnus.backends.ckan.CKANBackendImpl.insert(CKANBackendImpl.java:185), com.telefonica.iot.cygnus.backends.ckan.CKANBackendImpl.persist(CKANBackendImpl.java:90), com.telefonica.iot.cygnus.sinks.NGSICKANSink.persistAggregation(NGSICKANSink.java:361), com.telefonica.iot.cygnus.sinks.NGSICKANSink.persistBatch(NGSICKANSink.java:274), com.telefonica.iot.cygnus.sinks.NGSISink.processRollbackedBatches(NGSISink.java:395), com.telefonica.iot.cygnus.sinks.NGSISink.process(NGSISink.java:373), org.apache.flume.sink.DefaultSinkProcessor.process(DefaultSinkProcessor.java:67), org.apache.flume.SinkRunner$PollingRunner.run(SinkRunner.java:145), java.lang.Thread.run(Thread.java:748)]
'''''

[fgalan]

It seems that the request to CKAN API results in error:

CygnusPersistenceError. Could not insert (resId=46bcc158-ca18-49c9-b999-19aa30b46452, statusCode=409)

Maybe the Cygnus log trace could be prettier and provide more information, but the cause seems to be at CKAN. I don't know why CKAN is responding with 409 response code.

Maybe getting traces of the request sent by Cygnus and response received by CKAN could help. I'd suggest to increase Cygnus log level to DEBUG and check CKAN logs also. If that doesn't provide useful information, need step is to capture (using tcpdump or similar) and analyze the request-response between Cygnus and CKAN.

[amfgomez1219]

time=2021-02-17T14:52:07.560Z | lvl=DEBUG | corr=8e479502-712f-11eb-927f-26d4d8210cdb | trans=48b51fe9-8ba2-4579-b86e-8d18d3326d4f | srv=N/A | subsrv=N/A | comp=cygnus-ngsi | op=createJsonResponse | msg=com.telefonica.iot.cygnus.backends.http.HttpBackend[324] : Http response status line: HTTP/1.1 409 CONFLICT
time=2021-02-17T14:52:07.561Z | lvl=DEBUG | corr=8e479502-712f-11eb-927f-26d4d8210cdb | trans=48b51fe9-8ba2-4579-b86e-8d18d3326d4f | srv=N/A | subsrv=N/A | comp=cygnus-ngsi | op=createJsonResponse | msg=com.telefonica.iot.cygnus.backends.http.HttpBackend[360] : Http response payload: {"help": "https://catalogo.xxxxxxxx.es/api/3/action/help_show?name=datastore_upsert", "success": false, "error": {"records": ["row \"1\" has extra keys \"project\""], "__type": "Validation Error"}}
time=2021-02-17T14:52:07.561Z | lvl=ERROR | corr=8e479502-712f-11eb-927f-26d4d8210cdb | trans=48b51fe9-8ba2-4579-b86e-8d18d3326d4f | srv=N/A | subsrv=N/A | comp=cygnus-ngsi | op=processNewBatches | msg=com.telefonica.iot.cygnus.sinks.NGSISink[612] : CygnusPersistenceError. Could not insert (resId=47a86bc6-4ce9-4723-b4f0-2bde3f345dc1, statusCode=409). Stack trace: [com.telefonica.iot.cygnus.backends.ckan.CKANBackendImpl.insert(CKANBackendImpl.java:261), com.telefonica.iot.cygnus.backends.ckan.CKANBackendImpl.persist(CKANBackendImpl.java:99), com.telefonica.iot.cygnus.sinks.NGSICKANSink.persistAggregation(NGSICKANSink.java:378), com.telefonica.iot.cygnus.sinks.NGSICKANSink.persistBatch(NGSICKANSink.java:289), com.telefonica.iot.cygnus.sinks.NGSISink.processNewBatches(NGSISink.java:596), com.telefonica.iot.cygnus.sinks.NGSISink.process(NGSISink.java:376), org.apache.flume.sink.DefaultSinkProcessor.process(DefaultSinkProcessor.java:67), org.apache.flume.SinkRunner$PollingRunner.run(SinkRunner.java:145), java.lang.Thread.run(Thread.java:748)]`

[amfgomez1219]

time=2021-02-18T07:30:46.523Z | lvl=ERROR | corr=N/A | trans=N/A | srv=N/A | subsrv=N/A | comp=cygnus-ngsi | op=processRollbackedBatches | msg=com.telefonica.iot.cygnus.sinks.NGSISink[418] : CygnusPersistenceError. Could not insert (resId=b9a073a9-d062-4a82-8e9f-51b15eb85e8c, statusCode=409). Stack trace: [com.telefonica.iot.cygnus.backends.ckan.CKANBackendImpl.insert(CKANBackendImpl.java:261), com.telefonica.iot.cygnus.backends.ckan.CKANBackendImpl.persist(CKANBackendImpl.java:99), com.telefonica.iot.cygnus.sinks.NGSICKANSink.persistAggregation(NGSICKANSink.java:378), com.telefonica.iot.cygnus.sinks.NGSICKANSink.persistBatch(NGSICKANSink.java:289), com.telefonica.iot.cygnus.sinks.NGSISink.processRollbackedBatches(NGSISink.java:402), com.telefonica.iot.cygnus.sinks.NGSISink.process(NGSISink.java:375), org.apache.flume.sink.DefaultSinkProcessor.process(DefaultSinkProcessor.java:67), org.apache.flume.SinkRunner$PollingRunner.run(SinkRunner.java:145), java.lang.Thread.run(Thread.java:748)]
time=2021-02-18T07:30:46.523Z | lvl=INFO | corr=N/A | trans=N/A | srv=N/A | subsrv=N/A | comp=cygnus-ngsi | op=doRollbackAgain | msg=com.telefonica.iot.cygnus.sinks.NGSISink[493] : Rollbacking again (341433d8-71bb-11eb-b142-fe90fbf258e1), this was retry #1`

[amfgomez1219]

In CKAN I have data stored in CKAN resources. But the communication between cygnus-ckan return 409, if the entity does not change,
what could be the problem?

[fgalan]

From the logs:

... Http response status line: HTTP/1.1 409 CONFLICT
... Http response payload: {"help": "https://catalogo.xxxxxxxx.es/api/3/action/help_show?name=datastore_upsert", "success": false, "error": {"records": ["row \"1\" has extra keys \"project\""], "__type": "Validation Error"}}

So it seems there is some kind of validation error when Cygnus tries to interact with CKAN.

Which request is the one done by Cygnus that produces that error? If not easy to get that in the Cygnus logs, maybe capturing traffic in the Cygnus-CKAN interaction can help.

[sanamart]

Hello,

i would like to know if this issue was solved. Thank you.

[AlvaroVega]

Maybe could you tried with latest cygnus docker image and tell us, please?

[sanamart]

okay, i will try and i let you know.

[sanamart]

I still get the same error.

time=2022-05-08T11:32:59.906Z | lvl=INFO | corr=N/A | trans=N/A | srv=N/A | subsrv=N/A | comp=cygnus-ngsi | op=persistAggregation | msg=com.telefonica.iot.cygnus.sinks.NGSICKANSink[368] : [ckan-sink] Persisting data at NGSICKANSink (orgName=ci_test, pkgName=ci_test_testservicepath, resName=livingroomsensor_multisensor, data={"recvTime":"2022-05-08T11:31:40.12Z","fiwareServicePath":"/TestServicePath","entityId":"LivingRoomSensor","entityType":"multiSensor","Temperature":199.8,"Luminosity":186}) time=2022-05-08T11:32:59.950Z | lvl=ERROR | corr=N/A | trans=N/A | srv=N/A | subsrv=N/A | comp=cygnus-ngsi | op=processRollbackedBatches | msg=com.telefonica.iot.cygnus.sinks.NGSISink[418] : CygnusPersistenceError. Could not insert (resId=09c945b9-9f0a-4592-9567-a43db2ecc765, statusCode=404). Stack trace: [com.telefonica.iot.cygnus.backends.ckan.CKANBackendImpl.insert(CKANBackendImpl.java:261), com.telefonica.iot.cygnus.backends.ckan.CKANBackendImpl.persist(CKANBackendImpl.java:99), com.telefonica.iot.cygnus.sinks.NGSICKANSink.persistAggregation(NGSICKANSink.java:378), com.telefonica.iot.cygnus.sinks.NGSICKANSink.persistBatch(NGSICKANSink.java:289), com.telefonica.iot.cygnus.sinks.NGSISink.processRollbackedBatches(NGSISink.java:402), com.telefonica.iot.cygnus.sinks.NGSISink.process(NGSISink.java:375), org.apache.flume.sink.DefaultSinkProcessor.process(DefaultSinkProcessor.java:67), org.apache.flume.SinkRunner$PollingRunner.run(SinkRunner.java:145), java.lang.Thread.run(Thread.java:750)] time=2022-05-08T11:32:59.950Z | lvl=INFO | corr=N/A | trans=N/A | srv=N/A | subsrv=N/A | comp=cygnus-ngsi | op=doRollbackAgain | msg=com.telefonica.iot.cygnus.sinks.NGSISink[498] : Finishing internal transaction (6d8a4188-cec2-11ec-920b-6a6dc0ce0ddb), this was retry #10

I added a an Entity, that all worked without any problems. Then in CKAN i deleted the dataset and also any trace of it in the Admin-Trash. Then when i try to add again the same Entity i get the 404. If i restart Cygnus and try again, it works perfectly.

Do you think the error is more related to CKAN than to Cygnus implementation?

[sanamart]

Hello Alvaro,

I would like to know if you could give me some hints?

Thank you.

[AlvaroVega]

Hi @sanamart your previous reported errors where 409 (typically already created), now you are getting a 404 error (topically not found error). It seems not the same.

[sanamart]

Hi @AlvaroVega , should i open a new issue?