Feature/entity maps

CHANGES_NEXT_RELEASE

@@ -15,6 +15,8 @@ Fixed issues:
   * #1458 - Supporting system timestamps (createdAt/modifiedAt) in q
   * #1451 - Bug fix - removed a trailing ampersand from the URI for the connection to mongodb
   * #1418 - Performance - Faster startup when there are GeoProperties in DB
-  * #1478 - Fixed issue
+  * #1478 - Fixed issue about PATCH on non-existent entity in federation giving a 204 instead of a 404
   * #1479 - Fixed problem with "operations" in registrations
+  * #280  - Using Via header for loop detection (includes a new field 'hostAlias' in the registrations)
   * #280  - Fixed a bug in the lookup of pernot subscriptions
+  * #1496 - Fixed a crash - a double free()

CMakeLists.txt

@@ -236,6 +236,7 @@ SET (ORION_LIBS
     orionld_dbModel
     orionld_apiModel
     orionld_common
+    orionld_entityMaps
     orionld_types
     parse
     apiTypesV2
@@ -367,6 +368,7 @@ if (error EQUAL 0)
   ADD_SUBDIRECTORY(src/lib/jsonParse)
   ADD_SUBDIRECTORY(src/lib/jsonParseV2)
   ADD_SUBDIRECTORY(src/lib/rest)
+  ADD_SUBDIRECTORY(src/lib/orionld/entityMaps)
   ADD_SUBDIRECTORY(src/lib/orionld/pernot)
   ADD_SUBDIRECTORY(src/lib/orionld/socketService)
   ADD_SUBDIRECTORY(src/lib/orionld/notifications)

scripts/style_check_in_makefile.sh

@@ -46,16 +46,16 @@ function style_check
 }
 
 
+style_check src/lib/orionld/serviceRoutines
+style_check src/lib/orionld/forwarding
 style_check src/lib/orionld/apiModel
 style_check src/lib/orionld/regCache
-style_check src/lib/orionld/forwarding
 style_check src/lib/orionld/mongoc
 style_check src/lib/orionld/dbModel
 style_check src/lib/orionld/notifications
 style_check src/lib/orionld/common
 style_check src/lib/orionld/prometheus
 style_check src/lib/orionld/payloadCheck
-style_check src/lib/orionld/serviceRoutines
 style_check src/lib/orionld/legacyDriver
 style_check src/lib/orionld/q
 style_check src/lib/orionld/rest

src/app/orionld/orionld.cpp

@@ -125,13 +125,15 @@ extern "C"
 #include "orionld/context/orionldContextFromUrl.h"            // contextDownloadListInit, contextDownloadListRelease
 #include "orionld/contextCache/orionldContextCacheRelease.h"  // orionldContextCacheRelease
 #include "orionld/rest/orionldServiceInit.h"                  // orionldServiceInit
+#include "orionld/entityMaps/entityMapsRelease.h"             // entityMapsRelease
 #include "orionld/db/dbInit.h"                                // dbInit
 #include "orionld/mqtt/mqttRelease.h"                         // mqttRelease
 #include "orionld/regCache/regCacheInit.h"                    // regCacheInit
 #include "orionld/regCache/regCacheCreate.h"                  // regCacheCreate
 #include "orionld/regCache/regCacheRelease.h"                 // regCacheRelease
 #include "orionld/pernot/pernotSubCacheInit.h"                // pernotSubCacheInit
 #include "orionld/pernot/pernotLoop.h"                        // pernotLoopStart
+#include "orionld/pernot/pernotRelease.h"                     // pernotRelease
 
 #include "orionld/version.h"
 #include "orionld/orionRestServices.h"
@@ -147,9 +149,6 @@ extern "C"
 #include "orionld/troe/pgConnectionPoolsFree.h"               // pgConnectionPoolsFree
 #include "orionld/troe/pgConnectionPoolsPresent.h"            // pgConnectionPoolsPresent
 #include "orionld/forwarding/distOpInit.h"                    // distOpInit
-#include "orionld/socketService/socketServiceInit.h"          // socketServiceInit
-#include "orionld/socketService/socketServiceRun.h"           // socketServiceRun
-
 
 #include "orionld/version.h"
 #include "orionld/orionRestServices.h"
@@ -250,6 +249,8 @@ int             troePoolSize;
 bool            socketService;
 unsigned short  socketServicePort;
 bool            distributed;
+char            brokerId[136];
+char            wip[512];
 bool            noNotifyFalseUpdate;
 bool            idIndex;
 bool            noswap;
@@ -334,6 +335,8 @@ bool            triggerOperation = false;
 #define SOCKET_SERVICE_DESC    "enable the socket service - accept connections via a normal TCP socket"
 #define SOCKET_SERVICE_PORT_DESC  "port to receive new socket service connections"
 #define DISTRIBUTED_DESC       "turn on distributed operation"
+#define BROKER_ID_DESC         "identity of this broker instance for registrations - for the Via header"
+#define WIP_DESC               "Enable concepts that are 'Work In Progress' (e.g. -wip entityMaps)"
 #define FORWARDING_DESC        "turn on distributed operation (deprecated)"
 #define ID_INDEX_DESC          "automatic mongo index on _id.id"
 #define NOSWAP_DESC            "no swapping - for testing only!!!"
@@ -348,7 +351,7 @@ bool            triggerOperation = false;
 #define DBURI_DESC             "complete URI for database connection"
 #define DEBUG_CURL_DESC        "turn on debugging of libcurl - to the broker's logfile"
 #define CSUBCOUNTERS_DESC      "number of subscription counter updates before flush from sub-cache to DB (0: never, 1: always)"
-#define CORE_CONTEXT_DESC      "Core context version (v1.0|v1.3|v1.4|v1.5|v1.6|v1.7) - v1.6 is default"
+#define CORE_CONTEXT_DESC      "core context version (v1.0|v1.3|v1.4|v1.5|v1.6|v1.7) - v1.6 is default"
 
 
 
@@ -435,12 +438,14 @@ PaArgument paArgs[] =
   { "-troeUser",              troeUser,                 "TROE_USER",                 PaString,  PaOpt,  _i "postgres",   PaNL,   PaNL,             TROE_HOST_USER           },
   { "-troePwd",               troePwd,                  "TROE_PWD",                  PaString,  PaOpt,  _i "password",   PaNL,   PaNL,             TROE_HOST_PWD            },
   { "-troePoolSize",          &troePoolSize,            "TROE_POOL_SIZE",            PaInt,     PaOpt,  10,              0,      1000,             TROE_POOL_DESC           },
-  { "-distributed",           &distributed,             "DISTRIBUTED",               PaBool,    PaOpt,  false,           false,  true,             DISTRIBUTED_DESC         },
   { "-noNotifyFalseUpdate",   &noNotifyFalseUpdate,     "NO_NOTIFY_FALSE_UPDATE",    PaBool,    PaOpt,  false,           false,  true,             NO_NOTIFY_FALSE_UPDATE_DESC  },
-  { "-triggerOperation",      &triggerOperation,        "TRIGGER_OPERATION",         PaBool,    PaHid,  false,           false,  true,             TRIGGER_OPERATION_DESC   },
   { "-experimental",          &experimental,            "EXPERIMENTAL",              PaBool,    PaOpt,  false,           false,  true,             EXPERIMENTAL_DESC        },
   { "-mongocOnly",            &mongocOnly,              "MONGOCONLY",                PaBool,    PaOpt,  false,           false,  true,             MONGOCONLY_DESC          },
   { "-cSubCounters",          &cSubCounters,            "CSUB_COUNTERS",             PaInt,     PaOpt,  20,              0,      PaNL,             CSUBCOUNTERS_DESC        },
+  { "-distributed",           &distributed,             "DISTRIBUTED",               PaBool,    PaOpt,  false,           false,  true,             DISTRIBUTED_DESC         },
+  { "-brokerId",              &brokerId,                "BROKER_ID",                 PaStr,     PaOpt,  _i "",           PaNL,   PaNL,             BROKER_ID_DESC           },
+  { "-wip",                   wip,                      "WIP",                       PaStr,     PaHid,  _i "",           PaNL,   PaNL,             WIP_DESC                 },
+  { "-triggerOperation",      &triggerOperation,        "TRIGGER_OPERATION",         PaBool,    PaHid,  false,           false,  true,             TRIGGER_OPERATION_DESC   },
   { "-forwarding",            &distributed,             "FORWARDING",                PaBool,    PaHid,  false,           false,  true,             FORWARDING_DESC          },
   { "-socketService",         &socketService,           "SOCKET_SERVICE",            PaBool,    PaHid,  false,           false,  true,             SOCKET_SERVICE_DESC      },
   { "-ssPort",                &socketServicePort,       "SOCKET_SERVICE_PORT",       PaUShort,  PaHid,  1027,            PaNL,   PaNL,             SOCKET_SERVICE_PORT_DESC },
@@ -630,6 +635,14 @@ void exitFunc(void)
     pgConnectionPoolsFree();
   }
 
+  // Cleanup entity maps
+  if (entityMaps != NULL)
+    entityMapsRelease();
+
+  // Cleanup periodic notifications
+  if (pernot == true)
+    pernotRelease();
+
   kaBufferReset(&kalloc, KFALSE);
 }
 
@@ -926,7 +939,7 @@ static char* coreContextUrlSetup(const char* version)
 */
 int main(int argC, char* argV[])
 {
-# if 0
+#if 0
   //
   // Just an experiment.
   // It's an interesting way of "comparing strings"
@@ -1039,7 +1052,7 @@ int main(int argC, char* argV[])
   }
 
   //
-  // If trace levels are set, turn set logLevel to DEBUG, so that the trace messages will actually pass through
+  // If trace levels are set, set logLevel to DEBUG, so that the trace messages will actually pass through
   //
   if (paIsSet(argC, argV, paArgs, "-t"))
     strncpy(paLogLevel, "DEBUG", sizeof(paLogLevel) - 1);
@@ -1052,6 +1065,18 @@ int main(int argC, char* argV[])
 
   lmTimeFormat(0, (char*) "%Y-%m-%dT%H:%M:%S");
 
+  if ((debugCurl == true) && ((lmTraceIsSet(LmtCurl) == false) || (strcmp(paLogLevel, "DEBUG") != 0)))
+  {
+    strncpy(paLogLevel, "DEBUG", sizeof(paLogLevel) - 1);
+    lmTraceLevelSet(LmtCurl, true);
+  }
+
+  if (wip[0] != 0)
+  {
+    if (strcmp(wip, "entityMaps") == 0)
+      entityMapsEnabled = true;
+  }
+
 #if 0
   //
   // Uncomment this piece of code and run the functests (-ld) to make sure everything works "more or less" with the Legacy Driver disabled.
@@ -1097,14 +1122,10 @@ int main(int argC, char* argV[])
   paCleanup();
 
   if (strlen(dbName) > DB_NAME_MAX_LEN)
-  {
     LM_X(1, ("dbName too long (max %d characters)", DB_NAME_MAX_LEN));
-  }
 
   if (useOnlyIPv6 && useOnlyIPv4)
-  {
     LM_X(1, ("Fatal Error (-ipv4 and -ipv6 can not be activated at the same time. They are incompatible)"));
-  }
 
   if (https)
   {
@@ -1160,7 +1181,6 @@ int main(int argC, char* argV[])
   contextDownloadListInit();
 
 
-
   //
   // Initialize the KBASE library
   // This call redirects all log messages from the K-libs to the brokers log file.
@@ -1191,6 +1211,12 @@ int main(int argC, char* argV[])
   // localIpAndPort - IP:port for X-Forwarded-For
   snprintf(localIpAndPort, sizeof(localIpAndPort), "%s:%d", orionldHostName, port);
 
+  // brokerId - for the Via header
+  if (brokerId[0] == 0)
+    strncpy(brokerId, localIpAndPort, sizeof(brokerId) - 1);
+  else
+    distributed = true;  // Turn on forwarding if the brokerId CLI is used
+
   orionldStateInit(NULL);
 
   // mongocInit calls mongocGeoIndexInit - tenant0 must be ready for that

src/app/orionld/orionldRestServices.cpp

@@ -65,6 +65,8 @@
 #include "orionld/serviceRoutines/orionldOptions.h"
 #include "orionld/serviceRoutines/orionldPutEntity.h"
 #include "orionld/serviceRoutines/orionldPutAttribute.h"
+#include "orionld/serviceRoutines/orionldGetEntityMap.h"
+#include "orionld/serviceRoutines/orionldDeleteEntityMap.h"
 
 #include "orionld/serviceRoutines/orionldGetTemporalEntities.h"
 #include "orionld/serviceRoutines/orionldGetTemporalEntity.h"
@@ -90,6 +92,7 @@ static OrionLdRestServiceSimplified getServiceV[] =
   { "/ngsi-ld/ex/v1/ping",                 orionldGetPing             },
   { "/ngsi-ld/v1/entities/*",              orionldGetEntity           },
   { "/ngsi-ld/v1/entities",                orionldGetEntities         },
+  { "/ngsi-ld/v1/entityMaps/*",            orionldGetEntityMap        },
   { "/ngsi-ld/v1/types/*",                 orionldGetEntityType       },
   { "/ngsi-ld/v1/types",                   orionldGetEntityTypes      },
   { "/ngsi-ld/v1/attributes/*",            orionldGetEntityAttribute  },
@@ -173,6 +176,7 @@ static OrionLdRestServiceSimplified deleteServiceV[] =
 {
   { "/ngsi-ld/v1/entities/*/attrs/*",             orionldDeleteAttribute                 },
   { "/ngsi-ld/v1/entities/*",                     orionldDeleteEntity                    },
+  { "/ngsi-ld/v1/entityMaps/*",                   orionldDeleteEntityMap                 },
   { "/ngsi-ld/v1/subscriptions/*",                orionldDeleteSubscription              },
   { "/ngsi-ld/v1/csourceRegistrations/*",         orionldDeleteRegistration              },
   { "/ngsi-ld/v1/jsonldContexts/*",               orionldDeleteContext                   },

src/lib/logMsg/traceLevels.h

@@ -72,54 +72,76 @@ typedef enum TraceLevels
   LmtRegCache = 60,                    // Registration Cache
 
   //
-  // Distributed Operations
+  // Distributed Operations - requests
   //
-  LmtDistOpMsgs = 70,                  // Distributed Operations: messages
-  LmtDistOpRequest,                    // ONLY the verb, path, and body of a distributed request
-  LmtDistOpResponse,                   // ONLY the body and status code of the response to a distributed request
-  LmtDistOp207,                        // Merging of the final 207 response
+  LmtDistOpRequest = 70,               // ONLY the verb, path, and body of a distributed request
+  LmtDistOpRequestHeaders,             // HTTP headers of distributed requests
+  LmtDistOpRequestParams,              // URL parameters of distributed requests
+
+  //
+  // Distributed Operations - responses
+  //
+  LmtDistOpResponse = 80,              // ONLY the body and status code of the response to a distributed request
   LmtDistOpResponseBuf,                // Specific debugging of the incoming response of a distributed message
   LmtDistOpResponseDetail,             // Details on responses to distributed requests
   LmtDistOpResponseHeaders,            // HTTP headers of responses to distributed requests
-  LmtDistOpRequestHeaders,             // HTTP headers of request of distributed requests
-  LmtDistOpList,                       // Linked list of DistOps
+
+  //
+  // Distributed Operations - misc
+  //
+  LmtDistOpList = 90,                  // Linked list of DistOps
+  LmtDistOpAttributes,                 // The union of attributes URL-Param / Registered Attributes
+  LmtDistOpMerge,                      // Merge of responses from forwsrded requests (GET /entities)
+  LmtDistOpLoop,                       // Loop detection in forwarded messages
+  LmtDistOp207,                        // Merging of the final 207 response
 
   //
   // Context
   //
-  LmtContexts = 80,                    // Contexts
+  LmtContexts = 100,                   // Contexts
   LmtContextTree,                      // Context Tree
   LmtContextCache,                     // Context Cache
   LmtContextDownload,                  // Context Download
   LmtCoreContext,                      // Core Context
 
   // GeoJSON
-  LmtGeoJSON = 90,                     // GeoJSON ... everything (for now)
+  LmtGeoJSON = 110,                    // GeoJSON ... everything (for now)
 
   //
   // Pernot sub-cache
   //
-  LmtPernot = 100,                     // Periodic Notification Subscription cache
+  LmtPernot = 120,                     // Periodic Notification Subscription cache
   LmtPernotLoop,                       // Pernot loop, when each sub is triggered in time
   LmtPernotLoopTimes,                  // Pernot loop, details on timestamps
   LmtPernotFlush,                      // Pernot flush to DB
   LmtPernotQuery,                      // Pernot query
 
+  //
+  // Pagination
+  //
+  LmtEntityMap = 130,                  // The arrays of registrations per entity - distributed GET /entities
+  LmtEntityMapRetrieve,                // Retrieval of an entity map
+  LmtEntityMapDetail,                  // Details of the entity-registration maps
+
   //
   // Misc
   //
-  LmtMongoc = 230,                     // Entire mongoc library
-  LmtSR,                               // Service Routine (whatever it is doing)
+  LmtMongoc = 200,                     // Entire mongoc library
+  LmtSR,                               // Service Routine (whatever it is it's doing)
+  LmtCount,                            // NGSILD-Results-Count header, details for distops
   LmtSemaphore,                        // Semaphores
   LmtKjlParse,                         // Trace level start for K libs
   LmtLegacy,                           // Old code (mongoBackend, json parsers, etc)
   LmtLegacySubMatch,                   // Old code - update/subscription match for subs/notifs
   LmtLegacySubCacheRefresh,            // Old code - sub-cache-refresh
   LmtMqtt,                             // MQTT notifications
-  LmtQ,                                // Query language
+  LmtQ,                                // Query Language
+  LmtPostgres,                         // Postgres (TRoE)
   LmtSql,                              // SQL command for TRoE
   LmtPgPool,                           // Postgres Connection Pool
+  LmtTenants,                          // Well, tenants :)
   LmtSocketService,                    // Socket Service
+  LmtRegex,                            // Regular expressions - all of them
 
   LmtCurl    = 250,                    // CURL library
   LmtToDo,                             // To Do list

src/lib/ngsiNotify/Notifier.cpp

@@ -95,7 +95,7 @@ void Notifier::sendNotifyContextRequest
     bool                             blackList
 )
 {
-  pthread_t                         tid;
+  pthread_t tid;
 
   std::vector<SenderThreadParams*>* paramsV = Notifier::buildSenderParams(ncrP,
                                                                           httpInfo,

src/lib/ngsiNotify/QueueWorkers.cpp

@@ -157,7 +157,7 @@ static void* workerFunc(void* pSyncQ)
 
         if (ngsildSubscription == false)
           subscriptionId = NULL;
-
+ 
         LM_T(LmtNotificationMsg, ("Sending HTTP Notification for subscription '%s'", params->subscriptionId.c_str()));
         r = httpRequestSendWithCurl(curl,
                                     params->ip,

src/lib/orionld/common/orionldState.cpp

@@ -105,9 +105,12 @@ char              pgPortString[16];
 char              mongoServerVersion[32];
 char              userAgentHeaderNoLF[64];     // "User-Agent: orionld/" + ORIONLD_VERSION - initialized in orionldServiceInit()
 char              hostHeaderNoLF[128];
-char              hostHeader[128];             // Host: xxx
+char              hostHeader[256];             // Host: xxx
 size_t            hostHeaderLen;
 PernotSubCache    pernotSubCache;
+EntityMap*        entityMaps        = NULL;    // Used by GET /entities in the distributed case, for pagination
+bool              entityMapsEnabled = false;
+
 
 
 //