chore: create new system_tag_set table for Cyndi migration

RHINENG-23538

Author: RostyslavKachan

Dockerfile

@@ -61,7 +61,7 @@ RUN mkdir -p /tmp/prometheus_multiproc && \
 ENV prometheus_multiproc_dir=/tmp/prometheus_multiproc
 
 # minimal schema required by application, used for waiting in services until DB migration is finished
-ENV MINIMAL_SCHEMA=154
+ENV MINIMAL_SCHEMA=158
 
 # Baked-in content for FedRAMP
 ARG STATIC_ASSETS=0

common/peewee_model.py

@@ -148,6 +148,18 @@ class Meta:
         table_name = "system_group_set"
 
 
+class SystemTagSet(BaseModel):
+    """system_tag_set table"""
+
+    tags = BinaryJSONField(null=False)
+    tags_checksum = TextField(null=False, unique=True)
+
+    class Meta:
+        """system_tag_set table metadata"""
+
+        table_name = "system_tag_set"
+
+
 class SystemPlatform(BaseModelRW):
     """system_platform table"""
 
@@ -176,6 +188,7 @@ class SystemPlatform(BaseModelRW):
     host_type = TextField(null=True)
     operating_system_id = ForeignKeyField(column_name="operating_system_id", model=OperatingSystem, field="id")
     group_set_id = ForeignKeyField(column_name="group_set_id", model=SystemGroupSet, field="id")
+    tag_set_id = ForeignKeyField(column_name="tag_set_id", model=SystemTagSet, field="id")
 
     class Meta:
         """system_platform table metadata"""

database/schema/upgrade_scripts/158-tag_set.sql

@@ -0,0 +1,14 @@
+CREATE TABLE IF NOT EXISTS system_tag_set (
+  id BIGSERIAL,
+  tags JSONB NOT NULL,
+  tags_checksum TEXT NOT NULL,
+  PRIMARY KEY (id),
+  UNIQUE (tags_checksum)
+) TABLESPACE pg_default;
+
+CREATE INDEX ON system_tag_set USING gin (tags jsonb_path_ops);
+
+
+ALTER TABLE system_platform ADD COLUMN tag_set_id BIGINT;
+ALTER TABLE system_platform ADD CONSTRAINT tag_set_id FOREIGN KEY (tag_set_id) REFERENCES system_tag_set (id);
+CREATE INDEX ON system_platform(tag_set_id);

database/schema/ve_db_postgresql.sql

@@ -17,7 +17,7 @@ CREATE TABLE IF NOT EXISTS db_version (
 ) TABLESPACE pg_default;
 
 -- set the schema version directly in the insert statement here!!
-INSERT INTO db_version (name, version) VALUES ('schema_version', 157);
+INSERT INTO db_version (name, version) VALUES ('schema_version', 158);
 -- INSERT INTO db_version (name, version) VALUES ('schema_version', :schema_version);
 
 
@@ -191,6 +191,16 @@ CREATE TABLE IF NOT EXISTS system_group_set (
 
 CREATE INDEX ON system_group_set USING gin (groups jsonb_path_ops);
 
+CREATE TABLE IF NOT EXISTS system_tag_set (
+  id BIGSERIAL,
+  tags JSONB NOT NULL,
+  tags_checksum TEXT NOT NULL,
+  PRIMARY KEY (id),
+  UNIQUE (tags_checksum)
+) TABLESPACE pg_default;
+
+CREATE INDEX ON system_tag_set USING gin (tags jsonb_path_ops);
+
 -- system_platform
 CREATE TABLE IF NOT EXISTS system_platform (
   id BIGSERIAL,
@@ -220,6 +230,7 @@ CREATE TABLE IF NOT EXISTS system_platform (
   rule_results JSONB,
   operating_system_id INT,
   group_set_id BIGINT,
+  tag_set_id BIGINT,
   UNIQUE (inventory_id),
   CONSTRAINT rh_account_id
     FOREIGN KEY (rh_account_id)
@@ -229,7 +240,10 @@ CREATE TABLE IF NOT EXISTS system_platform (
     REFERENCES operating_system (id),
   CONSTRAINT group_set_id
     FOREIGN KEY (group_set_id)
-    REFERENCES system_group_set (id)
+    REFERENCES system_group_set (id),
+  CONSTRAINT tag_set_id
+    FOREIGN KEY (tag_set_id)
+    REFERENCES system_tag_set (id)
 ) TABLESPACE pg_default;
 
 CREATE INDEX ON system_platform(rh_account_id);
@@ -248,6 +262,8 @@ CREATE INDEX ON system_platform(operating_system_id);
 
 CREATE INDEX ON system_platform(group_set_id);
 
+CREATE INDEX ON system_platform(tag_set_id);
+
 CREATE TRIGGER system_platform_set_first_reported
   BEFORE INSERT ON system_platform
   FOR EACH ROW EXECUTE PROCEDURE set_first_reported();

listener/inventory_processor.py

@@ -56,6 +56,7 @@ def __init__(self, *args, **kwargs):
         self.grouper = MQWriter(CFG.grouper_inventory_topic, partitioner=Partitioners.org_id_partitioner, loop=self.loop)
         self.repo_id_cache = {}
         self.group_set_cache = {}
+        self.tag_set_cache = {}
 
     async def init(self):
         """Async constructor"""
@@ -282,11 +283,47 @@ async def _db_system_group_set_lookup(self, conn, groups: list[dict]) -> tuple[s
             result = await cur.fetchone()
             return groups_checksum, result["id"]
 
+    async def _db_system_tag_set_lookup(self, conn, tags: list[dict]) -> tuple[str, int]:
+        tags_str = json.dumps(tags)
+        tags_checksum = hashlib.sha256(tags_str.encode("utf-8")).hexdigest()
+        if tags_checksum in self.tag_set_cache:
+            return tags_checksum, self.tag_set_cache[tags_checksum]
+
+        async with conn.cursor(row_factory=dict_row) as cur:
+            await cur.execute(
+                "SELECT id FROM system_tag_set WHERE tags_checksum = %s",
+                (tags_checksum,),
+            )
+            result = await cur.fetchone()
+            if result:
+                self.tag_set_cache[tags_checksum] = result["id"]
+                return tags_checksum, result["id"]
+
+            await cur.execute(
+                """
+                    INSERT INTO system_tag_set (tags, tags_checksum)
+                    VALUES (%s, %s) ON CONFLICT (tags_checksum) DO UPDATE SET
+                        tags = EXCLUDED.tags
+                    RETURNING id
+                """,
+                (
+                    tags_str,
+                    tags_checksum,
+                ),
+            )
+            result = await cur.fetchone()
+            return tags_checksum, result["id"]
+
     async def _commit_system_group_set_cache(self, groups_checksum: str, group_set_id: int) -> None:
         """Insert group_set_id to the cache after succesful commit"""
         if groups_checksum not in self.group_set_cache:
             self.group_set_cache[groups_checksum] = group_set_id
 
+    async def _commit_system_tag_set_cache(self, tags_checksum: str, tag_set_id: int) -> None:
+        """Insert tag_set_id to the cache after succesful commit"""
+        if tags_checksum not in self.tag_set_cache:
+            self.tag_set_cache[tags_checksum] = tag_set_id
+
     async def _db_import_system(self, conn, fields: dict, org_id: str, inventory_id: str) -> (ImportStatus, int):
         """Import system to system_platform table, update if exists"""
         rh_account_id, system_id = await self._db_account_system_lookup(conn, org_id, inventory_id)
@@ -299,16 +336,16 @@ async def _db_import_system(self, conn, fields: dict, org_id: str, inventory_id:
                 INSERT INTO system_platform
                 (rh_account_id, inventory_id, s3_url, vmaas_json, json_checksum, display_name,
                  stale_timestamp, stale_warning_timestamp, culled_timestamp,
-                 host_type, last_upload, stale, operating_system_id, group_set_id)
+                 host_type, last_upload, stale, operating_system_id, group_set_id, tag_set_id)
                 VALUES (%(rh_account_id)s, %(inventory_id)s, %(s3_url)s, %(vmaas_json)s, %(json_checksum)s, %(display_name)s,
                         %(stale_timestamp)s, %(stale_warning_timestamp)s, %(culled_timestamp)s,
-                        %(host_type)s, %(last_upload)s, 'F', %(operating_system_id)s, %(group_set_id)s)
+                        %(host_type)s, %(last_upload)s, 'F', %(operating_system_id)s, %(group_set_id)s, %(tag_set_id)s)
                 ON CONFLICT (inventory_id) DO UPDATE SET
                 rh_account_id = EXCLUDED.rh_account_id, inventory_id = EXCLUDED.inventory_id, s3_url = EXCLUDED.s3_url, vmaas_json = EXCLUDED.vmaas_json,
                 json_checksum = EXCLUDED.json_checksum, display_name = EXCLUDED.display_name, stale_timestamp = EXCLUDED.stale_timestamp,
                 stale_warning_timestamp = EXCLUDED.stale_warning_timestamp, culled_timestamp = EXCLUDED.culled_timestamp,
                 host_type = EXCLUDED.host_type, last_upload = EXCLUDED.last_upload, stale = EXCLUDED.stale, operating_system_id = EXCLUDED.operating_system_id,
-                group_set_id = EXCLUDED.group_set_id
+                group_set_id = EXCLUDED.group_set_id, tag_set_id = EXCLUDED.tag_set_id
                 RETURNING (xmax = 0) AS inserted, unchanged_since, last_evaluation, id, when_deleted
             """
         else:
@@ -325,7 +362,8 @@ async def _db_import_system(self, conn, fields: dict, org_id: str, inventory_id:
                     last_upload = %(last_upload)s,
                     stale = FALSE,
                     operating_system_id = %(operating_system_id)s,
-                    group_set_id = %(group_set_id)s
+                    group_set_id = %(group_set_id)s,
+                    tag_set_id = %(tag_set_id)s
                 WHERE id = %(system_id)s
                 RETURNING (xmax = 0) AS inserted, unchanged_since, last_evaluation, id, when_deleted
             """
@@ -439,8 +477,10 @@ async def _process_upload(self, msg: InventoryMsg):
                     operating_system = msg.msg["host"]["system_profile"]["operating_system"]
                     operating_system_id = await self._db_operating_system_lookup(conn, operating_system)
                     groups_checksum, group_set_id = await self._db_system_group_set_lookup(conn, msg.msg["host"]["groups"])
+                    tags_checksum, tag_set_id = await self._db_system_tag_set_lookup(conn, msg.msg["host"]["tags"])
                     insert_fields["operating_system_id"] = operating_system_id
                     insert_fields["group_set_id"] = group_set_id
+                    insert_fields["tag_set_id"] = tag_set_id
                     import_status, system_id = await self._db_import_system(conn, insert_fields, org_id, inventory_id)
                     if ImportStatus.FAILED in import_status:
                         return
@@ -454,6 +494,7 @@ async def _process_upload(self, msg: InventoryMsg):
                 # Insert new items to the cache only after succesfully commited insert/update
                 await self._commit_operating_system_cache(operating_system, operating_system_id)
                 await self._commit_system_group_set_cache(groups_checksum, group_set_id)
+                await self._commit_system_tag_set_cache(tags_checksum, tag_set_id)
             except Exception as exc:
                 DATABASE_ERROR.inc()
                 LOGGER.exception("Error importing system: %s", exc)

listener/listener.py

@@ -47,7 +47,7 @@ class InventoryMsgValidator:
     """Inventory kafka message validator"""
 
     _created_updated_msg_fields = {
-        "host": ["id", "org_id", "system_profile", "display_name", "reporter", "per_reporter_staleness", "groups"],
+        "host": ["id", "org_id", "system_profile", "display_name", "reporter", "per_reporter_staleness", "groups", "tags"],
         "timestamp": [],
         "type": [],
     }

tests/taskomatic_tests/test_db_metrics.py

@@ -21,4 +21,4 @@ def test_db_metrics(pg_db_conn, monkeypatch):  # pylint: disable=unused-argument
 
         assert dm.METRIC_SYSTEMS.collect()[0].samples[0].value == 33  # there are 33 systems in DB
         assert dm.METRIC_CYNDI_SYSTEMS.collect()[0].samples[0].value == 33  # there are also 33 systems syndicated
-        assert len(dm.METRIC_TABLE_SIZE.collect()[0].samples) == 547
+        assert len(dm.METRIC_TABLE_SIZE.collect()[0].samples) == 548  # number of tables in DB