Fiware Cygnus - Error: collection already exists when Cygnus receives notifications two or more times

Question

我正在使用配置了 STH 的 Fiware Cygnus。当 Cygnus 在任何实体中收到两个或多个更改通知时，它无法将新值发送到 STH。它会记录错误 collection already exists 并且不会保存更改。在第一个通知中，一切顺利。

我执行了以下步骤：

在 Orion 中创建对任何实体的通知
更改实体属性

第一次，Cygnus 运行良好，所以我再次更改属性，Cygnus 显示错误。

如何解决这个问题？

完整消息是：

* 我在容器中使用 cygnus docker，如文档中所述。

* 我在这里使用 mongodb-ip:27017 而不是 post 的真实 IP。

time=2017-03-08T11:51:05.164Z | lvl=ERROR |
corr=53d86140-03f5-11e7-a70e-080027f6529d |
trans=236416c2-776e-4cc0-91dc-29bca203ea2a | srv=red | subsrv=/red/red |
comp=cygnus-ngsi | op=processRollbackedBatches |msg=com.telefonica.iot.cygnus.sinks.NGSISink[394] :
Persistence error. Message: -, Command failed with error -1:
'collection already exists' on server <mongodb-ip>:27017. The full response is
{ "ok" : 0.0, "errmsg" : "collection already exists" }, Stack trace:
 [com.telefonica.iot.cygnus.sinks.NGSISTHSink.persistOne(NGSISTHSink.java:158),
com.telefonica.iot.cygnus.sinks.NGSISTHSink.persistBatch(NGSISTHSink.java:93), 
com.telefonica.iot.cygnus.sinks.NGSISink.processRollbackedBatches(NGSISink.java:387), 
com.telefonica.iot.cygnus.sinks.NGSISink.process(NGSISink.java:370),
org.apache.flume.sink.DefaultSinkProcessor.process(DefaultSinkProcessor.java:68), 
org.apache.flume.SinkRunner$PollingRunner.run(SinkRunner.java:147),
java.lang.Thread.run(Thread.java:745)]
time=2017-03-08T11:51:05.164Z | lvl=INFO | corr=53d86140-03f5-11e7-a70e-080027f6529d | trans=236416c2-776e-4cc0-91dc-29bca203ea2a | srv=red | subsrv=/red/red | comp=cygnus-ngsi | op=doRollbackAgain | msg=com.telefonica.iot.cygnus.sinks.NGSISink[464] : Finishing internal transaction (53d86140-03f5-11e7-a70e-080027f6529d), this was retry #10

我在 agent.conf (Cygnus) 中使用的配置是：

cygnus-ngsi.sources = http-source
cygnus-ngsi.sinks = sth-sink
cygnus-ngsi.channels = sth-channel

cygnus-ngsi.sources.http-source.type = org.apache.flume.source.http.HTTPSource
cygnus-ngsi.sources.http-source.channels = sth-channel
cygnus-ngsi.sources.http-source.port = 5050
cygnus-ngsi.sources.http-source.handler = com.telefonica.iot.cygnus.handlers.NGSIRestHandler
cygnus-ngsi.sources.http-source.handler.notification_target = /notify
cygnus-ngsi.sources.http-source.handler.default_service = default
cygnus-ngsi.sources.http-source.handler.default_service_path = /
cygnus-ngsi.sources.http-source.interceptors = ts gi
cygnus-ngsi.sources.http-source.interceptors.ts.type = timestamp
cygnus-ngsi.sources.http-source.interceptors.gi.type = com.telefonica.iot.cygnus.interceptors.NGSIGroupingInterceptor$Builder
cygnus-ngsi.sources.http-source.interceptors.gi.grouping_rules_conf_file = /opt/apache-flume/conf/grouping_rules.conf

cygnus-ngsi.sinks.sth-sink.type = com.telefonica.iot.cygnus.sinks.NGSISTHSink
cygnus-ngsi.sinks.sth-sink.channel = sth-channel
#cygnus-ngsi.sinks.sth-sink.enable_encoding = false
#cygnus-ngsi.sinks.sth-sink.enable_grouping = false
#cygnus-ngsi.sinks.sth-sink.enable_name_mappings = false
#cygnus-ngsi.sinks.sth-sink.enable_lowercase = false
cygnus-ngsi.sinks.sth-sink.data_model = dm-by-entity
cygnus-ngsi.sinks.sth-sink.mongo_hosts = <mongodb-ip>:27017
cygnus-ngsi.sinks.sth-sink.mongo_username =
cygnus-ngsi.sinks.sth-sink.mongo_password =
cygnus-ngsi.sinks.sth-sink.db_prefix = sth_
cygnus-ngsi.sinks.sth-sink.collection_prefix = sth_
cygnus-ngsi.sinks.sth-sink.resolutions = day,hour,minute
#cygnus-ngsi.sinks.sth-sink.batch_size = 1
#cygnus-ngsi.sinks.sth-sink.batch_timeout = 30
#cygnus-ngsi.sinks.sth-sink.batch_ttl = 10
#cygnus-ngsi.sinks.sth-sink.data_expiration = 0
#cygnus-ngsi.sinks.sth-sink.ignore_white_spaces = true


cygnus-ngsi.channels.sth-channel.type = com.telefonica.iot.cygnus.channels.CygnusMemoryChannel
cygnus-ngsi.channels.sth-channel.capacity = 1000
cygnus-ngsi.channels.sth-channel.transactionCapacity = 100

在 STH 方面我有以下配置：

var config = {};

// STH server configuration
//--------------------------
config.server = {

  host: '10.0.2.15',

  port: '8666',


  defaultService: 'testservice',


  defaultServicePath: '/testservicepath',


  filterOutEmpty: 'true',

  aggregationBy: ['day', 'hour', 'minute'],

  temporalDir: 'temp'
};

// Database configuration
//------------------------
config.database = {

  dataModel: 'collection-per-entity',
 user: '',

  password: '',

  URI: 'localhost:27017',

  replicaSet: '',

  prefix: 'sth_',


  collectionPrefix: 'sth_',

  poolSize: '5',


  shouldStore: 'both',
  truncation: {


    expireAfterSeconds: '0',
    size: '0',

    max: '0'
  },

  ignoreBlankSpaces: 'true',
  nameMapping: {

    enabled: 'false',

    configFile: './name-mapping.json'
  },

  nameEncoding: 'false'
};

// Logging configuration
//------------------------
config.logging = {
  level: 'info',
  NODE_ENV variable is set to 'development'.

  format: 'pipe',


  proofOfLifeInterval: '60'
};

module.exports = config;

Answer 1

MongoDB 错误处理基于可能因版本而异的异常消息。

Cygnus 已编码为检测这些异常消息中的字符串 "code" : 48，这意味着该集合已存在且未执行任何操作。 "Nothing is done" 表示没有创建集合（因为它已经存在）并且没有进行原始异常。

问题是 MongoDB 版本 returns 消息不包含 "code" : 48 字符串。在这种情况下，异常被假定为不同于 "collection already existent"，因此它会被处理。这就是你的情况。

Fiware Cygnus - Error: collection already exists when Cygnus receives notifications two or more times

Fiware Cygnus - Error: collection already exists when Cygnus receives notifications two or more times

fiware

fiware-cygnus

fiware-sth-comet