FOGL-8285: Added new columns in streams table to track last sent stats and audit

C/services/north/data_load.cpp

@@ -118,6 +118,8 @@ bool DataLoad::setDataSource(const string& source)
 				source.c_str(), m_name.c_str());
 		return false;
 	}
+	m_lastFetched = getLastSentId();
+	m_streamSent = getLastSentId();
 	return true;
 }
 
@@ -350,7 +352,7 @@ unsigned long DataLoad::getLastSentId()
 		if (row)
 		{
 			// Get column value
-			ResultSet::ColumnValue* theVal = row->getColumn("last_object");
+			ResultSet::ColumnValue* theVal = row->getColumn(getStatsColumnName());
 			// Set found id
 			unsigned long rval = (unsigned long)theVal->getInteger();
 			delete lastObjectId;
@@ -458,7 +460,7 @@ int streamId = 0;
 InsertValues streamValues;
 
 	streamValues.push_back(InsertValue("description",    m_name));
-	streamValues.push_back(InsertValue("last_object",    0));
+	streamValues.push_back(InsertValue(getStatsColumnName(),    0));
 
 	if (m_storage->insertTable("streams", streamValues) != 1)
 	{
@@ -516,7 +518,7 @@ void DataLoad::flushLastSentId()
 	Where where("id", condition, to_string(m_streamId));
 	InsertValues lastId;
 
-	lastId.push_back(InsertValue("last_object", (long)m_streamSent));
+	lastId.push_back(InsertValue(getStatsColumnName(), (long)m_streamSent));
 	m_storage->updateTable("streams", lastId, where);
 }
 
@@ -725,3 +727,22 @@ void DataLoad::configChange(const string& category, const string& newConfig)
 		}
 	}
 }
+
+/**
+ * Get the stats column name to be used to fetch last reading sent with this current data source
+ */
+
+std::string	DataLoad::getStatsColumnName()
+{
+	std::string lastColName = "last_object";
+	switch(m_dataSource)
+	{
+		case SourceStatistics:
+			lastColName = "stats_last_object";
+			break;
+		case SourceAudit:
+			lastColName = "audit_last_object";
+			break;
+	}
+	return lastColName;
+}

C/services/north/include/data_load.h

@@ -70,6 +70,7 @@ class DataLoad : public ServiceHandler {
 		ReadingSet		*fetchAudit(unsigned int blockSize);
 		void			bufferReadings(ReadingSet *readings);
 		bool			loadFilters(const std::string& category);
+		std::string		getStatsColumnName();
 
 	private:
 		const std::string&	m_name;

VERSION

@@ -1,2 +1,2 @@
 fledge_version=2.6.0
-fledge_schema=74
+fledge_schema=75

scripts/plugins/storage/postgres/downgrade/74.sql

@@ -0,0 +1,2 @@
+ALTER TABLE fledge.streams DROP COLUMN stats_last_object;
+ALTER TABLE fledge.streams DROP COLUMN audit_last_object;

scripts/plugins/storage/postgres/init.sql

@@ -423,15 +423,17 @@ CREATE INDEX readings_ix3
 -- Streams table
 -- List of the streams to the Cloud.
 CREATE TABLE fledge.streams (
-       id             integer                     NOT NULL DEFAULT nextval('fledge.streams_id_seq'::regclass),         -- Sequence ID
-       description    character varying(255)      NOT NULL DEFAULT ''::character varying COLLATE pg_catalog."default",  -- A brief description of the stream entry
-       properties     jsonb                       NOT NULL DEFAULT '{}'::jsonb,                                         -- A generic set of properties
-       object_stream  jsonb                       NOT NULL DEFAULT '{}'::jsonb,                                         -- Definition of what must be streamed
-       object_block   jsonb                       NOT NULL DEFAULT '{}'::jsonb,                                         -- Definition of how the stream must be organised
-       object_filter  jsonb                       NOT NULL DEFAULT '{}'::jsonb,                                         -- Any filter involved in selecting the data to stream
-       active_window  jsonb                       NOT NULL DEFAULT '{}'::jsonb,                                         -- The window of operations
-       active         boolean                     NOT NULL DEFAULT true,                                                -- When false, all data to this stream stop and are inactive
-       last_object    bigint                      NOT NULL DEFAULT 0,                                                   -- The ID of the last object streamed (asset or reading, depending on the object_stream)
+       id                   integer                     NOT NULL DEFAULT nextval('fledge.streams_id_seq'::regclass),         -- Sequence ID
+       description          character varying(255)      NOT NULL DEFAULT ''::character varying COLLATE pg_catalog."default",  -- A brief description of the stream entry
+       properties           jsonb                       NOT NULL DEFAULT '{}'::jsonb,                                         -- A generic set of properties
+       object_stream        jsonb                       NOT NULL DEFAULT '{}'::jsonb,                                         -- Definition of what must be streamed
+       object_block         jsonb                       NOT NULL DEFAULT '{}'::jsonb,                                         -- Definition of how the stream must be organised
+       object_filter        jsonb                       NOT NULL DEFAULT '{}'::jsonb,                                         -- Any filter involved in selecting the data to stream
+       active_window        jsonb                       NOT NULL DEFAULT '{}'::jsonb,                                         -- The window of operations
+       active               boolean                     NOT NULL DEFAULT true,                                                -- When false, all data to this stream stop and are inactive
+       last_object          bigint                      NOT NULL DEFAULT 0,                                                   -- The ID of the last object streamed (asset or reading, depending on the object_stream)
+       stats_last_object    bigint                      NOT NULL DEFAULT 0,                                                   -- The ID of the last object streamed (asset or reading, depending on the object_stream)
+       audit_last_object    bigint                      NOT NULL DEFAULT 0,                                                   -- The ID of the last object streamed (asset or reading, depending on the object_stream)
        ts             timestamp(6) with time zone NOT NULL DEFAULT now(),                                               -- Creation or last update
        CONSTRAINT strerams_pkey PRIMARY KEY (id));
 

scripts/plugins/storage/postgres/upgrade/75.sql

@@ -0,0 +1,2 @@
+ALTER TABLE fledge.streams ADD COLUMN stats_last_object bigint DEFAULT 0;
+ALTER TABLE fledge.streams ADD COLUMN audit_last_object bigint DEFAULT 0;

scripts/plugins/storage/sqlite/downgrade/74.sql

@@ -0,0 +1,2 @@
+ALTER TABLE fledge.streams DROP COLUMN stats_last_object;
+ALTER TABLE fledge.streams DROP COLUMN audit_last_object;

scripts/plugins/storage/sqlite/init.sql

@@ -212,15 +212,17 @@ CREATE INDEX fki_asset_messages_fk2
 -- Streams table
 -- List of the streams to the Cloud.
 CREATE TABLE fledge.streams (
-    id            INTEGER                      PRIMARY KEY AUTOINCREMENT,         -- Sequence ID
-    description    character varying(255)      NOT NULL DEFAULT '',               -- A brief description of the stream entry
-    properties     JSON                        NOT NULL DEFAULT '{}',             -- A generic set of properties
-    object_stream  JSON                        NOT NULL DEFAULT '{}',             -- Definition of what must be streamed
-    object_block   JSON                        NOT NULL DEFAULT '{}',             -- Definition of how the stream must be organised
-    object_filter  JSON                        NOT NULL DEFAULT '{}',             -- Any filter involved in selecting the data to stream
-    active_window  JSON                        NOT NULL DEFAULT '{}',             -- The window of operations
-    active         boolean                     NOT NULL DEFAULT 't',              -- When false, all data to this stream stop and are inactive
-    last_object    bigint                      NOT NULL DEFAULT 0,                -- The ID of the last object streamed (asset or reading, depending on the object_stream)
+    id                  INTEGER                      PRIMARY KEY AUTOINCREMENT,         -- Sequence ID
+    description         character varying(255)      NOT NULL DEFAULT '',               -- A brief description of the stream entry
+    properties          JSON                        NOT NULL DEFAULT '{}',             -- A generic set of properties
+    object_stream       JSON                        NOT NULL DEFAULT '{}',             -- Definition of what must be streamed
+    object_block        JSON                        NOT NULL DEFAULT '{}',             -- Definition of how the stream must be organised
+    object_filter       JSON                        NOT NULL DEFAULT '{}',             -- Any filter involved in selecting the data to stream
+    active_window       JSON                        NOT NULL DEFAULT '{}',             -- The window of operations
+    active              boolean                     NOT NULL DEFAULT 't',              -- When false, all data to this stream stop and are inactive
+    last_object         bigint                      NOT NULL DEFAULT 0,                -- The ID of the last object streamed (asset or reading, depending on the object_stream)
+    stats_last_object   bigint                      NOT NULL DEFAULT 0,                -- The ID of the last object streamed (Statistics, depending on the object_stream)
+    audit_last_object   bigint                      NOT NULL DEFAULT 0,                -- The ID of the last object streamed (Audit, depending on the object_stream)
     ts             DATETIME DEFAULT (STRFTIME('%Y-%m-%d %H:%M:%f', 'NOW', 'localtime'))); -- Creation or last update
 
 

scripts/plugins/storage/sqlite/upgrade/75.sql

@@ -0,0 +1,2 @@
+ALTER TABLE fledge.streams ADD COLUMN stats_last_object bigint DEFAULT 0;
+ALTER TABLE fledge.streams ADD COLUMN audit_last_object bigint DEFAULT 0;

scripts/plugins/storage/sqlitelb/downgrade/74.sql

@@ -0,0 +1,2 @@
+ALTER TABLE fledge.streams DROP COLUMN stats_last_object;
+ALTER TABLE fledge.streams DROP COLUMN audit_last_object;

scripts/plugins/storage/sqlitelb/init.sql

@@ -212,15 +212,17 @@ CREATE INDEX fki_asset_messages_fk2
 -- Streams table
 -- List of the streams to the Cloud.
 CREATE TABLE fledge.streams (
-    id            INTEGER                      PRIMARY KEY AUTOINCREMENT,         -- Sequence ID
-    description    character varying(255)      NOT NULL DEFAULT '',               -- A brief description of the stream entry
-    properties     JSON                        NOT NULL DEFAULT '{}',             -- A generic set of properties
-    object_stream  JSON                        NOT NULL DEFAULT '{}',             -- Definition of what must be streamed
-    object_block   JSON                        NOT NULL DEFAULT '{}',             -- Definition of how the stream must be organised
-    object_filter  JSON                        NOT NULL DEFAULT '{}',             -- Any filter involved in selecting the data to stream
-    active_window  JSON                        NOT NULL DEFAULT '{}',             -- The window of operations
-    active         boolean                     NOT NULL DEFAULT 't',              -- When false, all data to this stream stop and are inactive
-    last_object    bigint                      NOT NULL DEFAULT 0,                -- The ID of the last object streamed (asset or reading, depending on the object_stream)
+    id                      INTEGER                      PRIMARY KEY AUTOINCREMENT,         -- Sequence ID
+    description             character varying(255)      NOT NULL DEFAULT '',               -- A brief description of the stream entry
+    properties              JSON                        NOT NULL DEFAULT '{}',             -- A generic set of properties
+    object_stream           JSON                        NOT NULL DEFAULT '{}',             -- Definition of what must be streamed
+    object_block            JSON                        NOT NULL DEFAULT '{}',             -- Definition of how the stream must be organised
+    object_filter           JSON                        NOT NULL DEFAULT '{}',             -- Any filter involved in selecting the data to stream
+    active_window           JSON                        NOT NULL DEFAULT '{}',             -- The window of operations
+    active                  boolean                     NOT NULL DEFAULT 't',              -- When false, all data to this stream stop and are inactive
+    last_object             bigint                      NOT NULL DEFAULT 0,                -- The ID of the last object streamed (asset or reading, depending on the object_stream)
+    stats_last_object       bigint                      NOT NULL DEFAULT 0,                -- The ID of the last object streamed (Statistics, depending on the object_stream)
+    audit_last_object       bigint                      NOT NULL DEFAULT 0,                -- The ID of the last object streamed (Audit, depending on the object_stream)
     ts             DATETIME DEFAULT (STRFTIME('%Y-%m-%d %H:%M:%f', 'NOW', 'localtime'))); -- Creation or last update
 
 

scripts/plugins/storage/sqlitelb/upgrade/75.sql

@@ -0,0 +1,2 @@
+ALTER TABLE fledge.streams ADD COLUMN stats_last_object bigint DEFAULT 0;
+ALTER TABLE fledge.streams ADD COLUMN audit_last_object bigint DEFAULT 0;