Process Ownership Collection

README.md

@@ -131,19 +131,20 @@ Once the configuration file is ready:
 ### Run the Installer
 
 To run the installer:
-1. Make sure you are authenticated. 
-2. Ensure Docker images are pushed to OCI.
-2. In a terminal or IDE, navigate to the DFA project's root. 
-3. Create and activate the virtual environment. 
-4. Install the following packages in the virtual environment using `pip install`
+1. Clone the DFA repository. 
+2. Make sure you are authenticated to OCI. 
+3. Ensure Docker images are pushed to OCI.
+4. In a terminal or IDE, navigate to the DFA project's root. 
+5. Create and activate the virtual environment. 
+6. Install the following packages in the virtual environment using `pip install`
     - fdk
     - oci
     - oracledb
     - pandas
     - pypika
-5. Run `export PYTHONPATH=/Users/yourName/oci-ag/src`
+7. Run `export PYTHONPATH=/Users/yourName/oci-ag/src`
 Paste the full path to DFA's source directory for the PYTHONPATH. 
-6. Run `python installer.py installer.log 2>&1`
+8. Run `python installer.py installer.log 2>&1`
 
 Note: The creation of the Vault and Master Key can take time, the script will wait a max of 20 minutes for an Active state before timing out. In the case of a timeout, please run the script again once the resource is in an active state in OCI. 
 

installer.py

@@ -27,6 +27,7 @@
 from dfa.adw.tables.cloud_policy import *
 from dfa.adw.tables.global_identity_collection import *
 from dfa.adw.tables.identity import *
+from dfa.adw.tables.ownership_collection import *
 from dfa.adw.tables.permission import *
 from dfa.adw.tables.permission_assignment import *
 from dfa.adw.tables.policy import *
@@ -69,6 +70,7 @@ def create_adw_tables():
     agrs_table = AccessGuardrailStateTable()
     aw_table = ApprovalWorkflowStateTable()
     aes_table = AuditEventsTable()
+    ocs_table = OwnershipCollectionStateTable()
 
     if os.environ.get("CREATE_TIME_SERIES", "false").lower() == "true":
         its_table = IdentityTimeSeriesTable()
@@ -84,6 +86,7 @@ def create_adw_tables():
         rolets_table = RoleTimeSeriesTable()
         agrts_table = AccessGuardrailTimeSeriesTable()
         awts_table = ApprovalWorkflowTimeSeriesTable()
+        octs_table = OwnershipCollectionTimeSeriesTable()
 
     ## Look for recreate environment variable to determine if DFA ADW tables need to be recreated (delete first)
     if "DFA_RECREATE_DFA_ADW_TABLES" in os.environ:
@@ -106,6 +109,7 @@ def create_adw_tables():
             agrs_table.delete()
             aw_table.delete()
             aes_table.delete()
+            ocs_table.delete()
 
             if os.environ.get("CREATE_TIME_SERIES", "false").lower() == "true":
                 its_table.delete()
@@ -121,6 +125,7 @@ def create_adw_tables():
                 rolets_table.delete()
                 agrts_table.delete()
                 awts_table.delete()
+                octs_table.delete()
 
     ## Create DFA ADW tables
     is_table.create()
@@ -137,6 +142,7 @@ def create_adw_tables():
     agrs_table.create()
     aw_table.create()
     aes_table.create()
+    ocs_table.create()
 
     if os.environ.get("CREATE_TIME_SERIES", "false").lower() == "true":
         its_table.create()
@@ -152,6 +158,7 @@ def create_adw_tables():
         rolets_table.create()
         agrts_table.create()
         awts_table.create()
+        octs_table.create()
 
 
 def setup(application_ocid):

src/dfa/adw/query_builders/ownership_collection.py

@@ -0,0 +1,133 @@
+# Copyright (c) 2025, Oracle and/or its affiliates.
+# Licensed under the Universal Permissive License v 1.0 as shown at https://oss.oracle.com/licenses/upl/.
+
+from abc import ABC, abstractmethod
+
+from pypika import Table
+
+from dfa.adw.connection import AdwConnection
+from dfa.adw.query_builders.base_query_builder import (
+    BaseQueryBuilder,
+    DeleteQueryBuilder,
+    InsertManyQueryBuilder,
+    UpdateManyQueryBuilder,
+)
+from dfa.adw.tables.ownership_collection import OwnershipCollectionStateTable, OwnershipCollectionTimeSeriesTable
+
+
+class OwnershipCollectionStateQueryBuilder(Table, ABC, BaseQueryBuilder):
+    table_manager = OwnershipCollectionStateTable()
+
+    def __init__(self, events: list):
+        super().__init__(self.table_manager.get_table_name().upper())
+        self.events = events
+
+    @abstractmethod
+    def execute_sql_for_events(self):
+        pass
+
+
+class OwnershipCollectionStateCreateQueryBuilder(OwnershipCollectionStateQueryBuilder):
+    def executemany_sql_for_events(self):
+        return OwnershipCollectionStateUpdateQueryBuilder(self.events).executemany_sql_for_events()
+
+    def execute_sql_for_events(self):
+        return self.executemany_sql_for_events()
+
+
+class OwnershipCollectionStateUpdateQueryBuilder(OwnershipCollectionStateQueryBuilder):
+    def executemany_sql_for_events(self):
+        self.logger.info(
+            "Using bulk insert / update operations for %d ownership collection events", len(self.events)
+        )
+
+        if len(self.events) == 0:
+            self.logger.info("No events to process by ownership collection query builder")
+            return
+
+        insert_statement = InsertManyQueryBuilder().get_operation_sql(self, self.events, [])
+        input_sizes = InsertManyQueryBuilder().get_input_sizes(self.events)
+        AdwConnection.get_cursor().setinputsizes(**input_sizes)
+        AdwConnection.get_cursor().executemany(insert_statement, self.events, batcherrors=True)
+
+        constraint_violating_rows = []
+        for batch_error in AdwConnection.get_cursor().getbatcherrors():
+            if batch_error.full_code == "ORA-00001":
+                constraint_violating_rows.append(self.events[batch_error.offset])
+            else:
+                self.logger.info("identity create failed - %s", batch_error.message)
+
+        if len(constraint_violating_rows) > 0:
+            self.logger.info(
+                "%d ownership collection creates failed for unique constraint violation - \
+                performing bulk ownership collection updates",
+                len(constraint_violating_rows),
+            )
+
+            update_sql = UpdateManyQueryBuilder().get_operation_sql(
+                self,
+                constraint_violating_rows,
+                [],
+                self.table_manager.get_unique_contraint_definition_details()["columns"],
+            )
+
+            AdwConnection.get_cursor().executemany(
+                update_sql, constraint_violating_rows, batcherrors=True
+            )
+
+            for batch_error in AdwConnection.get_cursor().getbatcherrors():
+                self.logger.info("ownership collection update failed - %s", batch_error.message)
+
+        unique_id_timstamp_pairs = DeleteQueryBuilder().remove_duplicates(self.events)
+        self.logger.info(
+            "Removing outdated rows for %d unique ownership collection id, timestamp pairs",
+            len(unique_id_timstamp_pairs),
+        )
+        for event in unique_id_timstamp_pairs:
+            delete_outdated_sql = DeleteQueryBuilder().delete_outdated_rows(self, event)
+            AdwConnection.get_cursor().execute(delete_outdated_sql)
+
+        AdwConnection.commit()
+
+    def execute_sql_for_events(self):
+        return self.executemany_sql_for_events()
+
+
+class OwnershipCollectionStateDeleteQueryBuilder(OwnershipCollectionStateQueryBuilder):
+    def execute_sql_for_events(self):
+        for event in self.events:
+            delete_sql = DeleteQueryBuilder().get_operation_sql(
+                self, event, ["id", "service_instance_id", "tenancy_id"]
+            )
+            AdwConnection.get_cursor().execute(delete_sql)
+            self.logger.info("Row delete for ownership collection delete request")
+
+        self.logger.info("Committing work for now")
+        AdwConnection.commit()
+
+
+class OwnershipCollectionTimeSeriesQueryBuilder(Table, ABC, BaseQueryBuilder):
+    table_manager = OwnershipCollectionTimeSeriesTable()
+
+    def __init__(self, events: list):
+        super().__init__(self.table_manager.get_table_name().upper())
+        self.events = events
+
+    @abstractmethod
+    def execute_sql_for_events(self):
+        pass
+
+
+class OwnershipCollectionTimeSeriesCreateQueryBuilder(OwnershipCollectionTimeSeriesQueryBuilder):
+    def execute_sql_for_events(self):
+        return self.executemany_sql_for_events()
+
+
+class OwnershipCollectionTimeSeriesUpdateQueryBuilder(OwnershipCollectionTimeSeriesQueryBuilder):
+    def execute_sql_for_events(self):
+        return self.executemany_sql_for_events()
+
+
+class OwnershipCollectionTimeSeriesDeleteQueryBuilder(OwnershipCollectionTimeSeriesQueryBuilder):
+    def execute_sql_for_events(self):
+        return self.executemany_sql_for_events()

src/dfa/adw/tables/ownership_collection.py

@@ -0,0 +1,41 @@
+# Copyright (c) 2025, Oracle and/or its affiliates.
+# Licensed under the Universal Permissive License v 1.0 as shown at https://oss.oracle.com/licenses/upl/.
+
+from dfa.adw.tables.base_table import BaseStateTable, BaseTable
+
+
+class OwnershipCollectionTimeSeriesTable(BaseTable):
+    _table_name = "ownership_collection_ts"
+    _schema = None
+
+    def _column_definitions(self):
+        json = """
+[
+    {"field_name":"ID","column_name":"ID","column_expression":null,"skip_column":false,"data_type":"VARCHAR2","data_length":32767,"data_format":null},
+    {"field_name":"ENTITY_ID","column_name":"ENTITY_ID","column_expression":null,"skip_column":false,"data_type":"VARCHAR2","data_length":32767,"data_format":null},
+    {"field_name":"ENTITY_NAME","column_name":"ENTITY_NAME","column_expression":null,"skip_column":false,"data_type":"VARCHAR2","data_length":32767,"data_format":null},
+    {"field_name":"IS_PRIMARY","column_name":"IS_PRIMARY","column_expression":null,"skip_column":false,"data_type":"VARCHAR2","data_length":32767,"data_format":null},
+    {"field_name":"EXTERNAL_ID","column_name":"EXTERNAL_ID","column_expression":null,"skip_column":false,"data_type":"VARCHAR2","data_length":32767,"data_format":null},
+    {"field_name":"RESOURCE_NAME","column_name":"RESOURCE_NAME","column_expression":null,"skip_column":false,"data_type":"VARCHAR2","data_length":32767,"data_format":null},
+    {"field_name":"CREATED_ON","column_name":"CREATED_ON","column_expression":null,"column_expression_type":null,"source_path":null,"source_column":null,"skip_column":false,"data_type":"NUMBER","data_length":null,"data_format":null},
+    {"field_name":"UPDATED_ON","column_name":"UPDATED_ON","column_expression":null,"column_expression_type":null,"source_path":null,"source_column":null,"skip_column":false,"data_type":"NUMBER","data_length":null,"data_format":null},    
+    {"field_name":"TENANCY_ID","column_name":"TENANCY_ID","column_expression":null,"skip_column":false,"data_type":"VARCHAR2","data_length":32767,"data_format":null},
+    {"field_name":"EVENT_OBJECT_TYPE","column_name":"EVENT_OBJECT_TYPE","column_expression":null,"skip_column":false,"data_type":"VARCHAR2","data_length":32767,"data_format":null},
+    {"field_name":"OPERATION_TYPE","column_name":"OPERATION_TYPE","column_expression":null,"skip_column":false,"data_type":"VARCHAR2","data_length":32767,"data_format":null},
+    {"field_name":"EVENT_TIMESTAMP","column_name":"EVENT_TIMESTAMP","column_expression":"SYSTIMESTAMP","skip_column":false,"data_type":"TIMESTAMP WITH TIME ZONE","data_length":null,"data_format":"YYYY-MM-DD\\"T\\"HH24:MI:SS.FFTZH:TZM"},
+    {"field_name":"ATTRIBUTES","column_name":"ATTRIBUTES","column_expression":null,"skip_column":false,"data_type":"CLOB","data_length":null,"data_format":null},
+    {"field_name":"SERVICE_INSTANCE_ID","column_name":"SERVICE_INSTANCE_ID","column_expression":null,"skip_column":false,"data_type":"VARCHAR2","data_length":32767,"data_format":null}
+]
+"""
+
+        return json
+
+
+class OwnershipCollectionStateTable(BaseStateTable, OwnershipCollectionTimeSeriesTable):
+    _table_name = "ownership_collection_state"
+
+    def get_unique_contraint_definition_details(self):
+        return {
+            "name": "DFA_UNQ_OC_ST_CONST",
+            "columns": ["ID", "ENTITY_ID", "SERVICE_INSTANCE_ID", "TENANCY_ID"],
+        }

src/dfa/etl/abstract_transformer.py

@@ -60,6 +60,7 @@ def is_valid_object_type(self, object_type):
             "ROLE",
             "ACCESS_GUARDRAIL",
             "APPROVAL_WORKFLOW",
+            "OWNERSHIP_COLLECTION"
         ]
 
         if object_type in valid_object_types:

src/dfa/etl/transformers/ownership_collection.py

@@ -0,0 +1,104 @@
+# Copyright (c) 2025, Oracle and/or its affiliates.
+# Licensed under the Universal Permissive License v 1.0 as shown at https://oss.oracle.com/licenses/upl/.
+
+import json
+
+from dfa.etl.transformers.base_event_transformer import BaseEventTransformer
+
+
+class OwnershipCollectionEventTransformer(BaseEventTransformer):
+    def transform_raw_event(self, raw_event):
+        ownership_collection = {
+            "id": "",
+            "entity_id": "",
+            "entity_name": "",
+            "is_primary": "", 
+            "external_id": "",
+            "resource_name": "",
+            "created_on": 0,
+            "updated_on": 0,
+            "tenancy_id": "",
+            "service_instance_id": "",
+            "event_object_type": "",
+            "operation_type": "",
+            "event_timestamp": "",
+            "attributes": "{}",
+        }
+
+        ownership_collection_list = []
+        try:
+            if self._get_tenancy_id():
+                ownership_collection["tenancy_id"] = self._get_tenancy_id()
+
+            if self._get_service_instance_id():
+                ownership_collection["service_instance_id"] = self._get_service_instance_id()
+
+            if self._get_event_timestamp():
+                ownership_collection["event_timestamp"] = self._get_event_timestamp()
+
+            if self.get_operation_type() == "DELETE":
+                if "id" in raw_event:
+                    ownership_collection["id"] = raw_event["id"]
+            else:
+                if "ownershipCollectionId" in raw_event:
+                    ownership_collection["id"] = raw_event["ownershipCollectionId"]
+
+            if "entityId" in raw_event:
+                ownership_collection["entity_id"] = raw_event["entityId"]
+
+            if "entityName" in raw_event:
+                ownership_collection["entity_name"] = raw_event["entityName"]
+
+            if "isPrimary" in raw_event:
+                ownership_collection["is_primary"] = raw_event["isPrimary"]
+
+            if "externalId" in raw_event:
+                ownership_collection["external_id"] = raw_event["externalId"]
+
+            if "usageName" in raw_event:
+                ownership_collection["resource_name"] = raw_event["usageName"]
+
+            if "timeCreated" in raw_event:
+                ownership_collection["created_on"] = raw_event["timeCreated"]
+
+            if "lastModified" in raw_event:
+                ownership_collection["updated_on"] = raw_event["lastModified"]
+
+            if "customAttributes" in raw_event:
+                ownership_collection["attributes"] = json.dumps(raw_event["customAttributes"])
+
+            ownership_collection["event_object_type"] = self.get_event_object_type()
+            ownership_collection["operation_type"] = self.get_operation_type()
+
+            ownership_collection_list.append(ownership_collection)
+
+        except KeyError as e:
+            self.logger.error(
+                "Cannot process event due to KeyError - %s is missing from event data", e
+            )
+
+        return ownership_collection_list
+
+    def transform_stream_message(self, message):
+        transformed_ownership_collections = []
+        # if isinstance(message.value['data'], list):
+        if isinstance(self._access_message_value_data(message), list):
+            # for event in message.value['data']:
+            for event in self._access_message_value_data(message):
+                transformed_ownership_collections.extend(self.transform_raw_event(event))
+        else:
+            transformed_ownership_collections.extend(super().transform_stream_message(message))
+
+        return transformed_ownership_collections
+
+
+class OwnershipCollectionCreateEventTransformer(OwnershipCollectionEventTransformer):
+    pass
+
+
+class OwnershipCollectionUpdateEventTransformer(OwnershipCollectionEventTransformer):
+    pass
+
+
+class OwnershipCollectionDeleteEventTransformer(OwnershipCollectionEventTransformer):
+    pass

tests/dfa/etl/test_data/file/ownership_collection.jsonl

@@ -0,0 +1,8 @@
+{"headers":{"eventId":"7458587f-9c95-430e-9ab2-2e31051a893b","correlationId":"e0993fdf-aad9-4e68-9a04-0304ac9f1852","eventTime":"2025-12-04T18:49:06.360205Z","eventType":"com.oracle.idm.agcs.data.enablement.ownershipCollection.created","eventTypeVersion":"1.0","operation":"CREATE","messageType":"OWNERSHIP_COLLECTION","status":"IN_PROGRESS","opcRequestId":"oci-80048803D6BD814-202512041828/1C35AA9BD84D1C0001E8D498ED38AAE7/04F668A6925253EFA2B21D8A07C859340K","tenancyId":"ocid1.tenancy.oc1..aaaaaaaazp2vvzjsn6newkqrpkwndxpdoixtqfgyhnf4y24h7d5ny2639054","serviceInstanceId":"ocid1.agcsgovernanceinstance.oc1.iad.amaaaaaaebkbezqaadpvwolr4raumlz3uxdgczwbqkalpcoo7qcu2r639054"}}
+{"ownershipCollectionId":"0030ffc6-a3a9-4aaa-9fbb-e487d47cc871","entityId":"globalId.OIG.18.w5rdiaax2ycwhy3g2ocmpc2nhhbmqtp57ue2nycsr7bzxpilla","isPrimary":"true","entityName":"Ama Maclead","externalId":"ocid1.agcsgovernanceinstance.oc1.iad.amaaaaaaebkbezqaadpvwolr4raumlz3uxdgczwbqkalpcoo7qcu2r639054","usageName":"dbumlowrisk","timeCreated":1741754088134,"lastModified":1741754088134}
+{"ownershipCollectionId":"0030ffc6-a3a9-4aaa-9fbb-e487d47cc871","entityId":"globalId.OIG.18.w5rdiaax2ycwhy3g2ocmpc2nhhbmqtp57ue2nycsr7bzxpilla","isPrimary":"false","entityName":"Tred Perlad","externalId":"ocid1.agcsgovernanceinstance.oc1.iad.amaaaaaaebkbezqaadpvwolr4raumlz3uxdgczwbqkalpcoo7qcu2r639054","usageName":"dbumlowrisk","timeCreated":1741754088134,"lastModified":1741754088134}
+{"ownershipCollectionId":"00599df6-fda4-48e4-b679-2cd870341f7e","entityId":"globalId.OIG.18.w5rdiaax2ycwhy3g2ocmpc2nhhbmqtp57ue2nycsr7bzxpilla","isPrimary":"true","entityName":"Maclead Ama","externalId":"ocid1.agcsgovernanceinstance.oc1.iad.amaaaaaaebkbezqaadpvwolr4raumlz3uxdgczwbqkalpcoo7qcu2r639054","usageName":"Analytics-User-Unity","timeCreated":1756734143175,"lastModified":1756734143175}
+{"ownershipCollectionId":"0067f1eb-143d-48fc-b503-e863c5df1e36","entityId":"globalId.ICF.Siebel_test6.f36cdc4c5c5cb6fb82b2cf7940fdfl85","isPrimary":"true","entityName":"AACCF NADLER","externalId":"ocid1.agcsgovernanceinstance.oc1.iad.amaaaaaaebkbezqaadpvwolr4raumlz3uxdgczwbqkalpcoo7qcu2r639054","usageName":"MYSQL_POLICY_AB","timeCreated":1730897026337,"lastModified":1730897026337}
+{"ownershipCollectionId":"006b5d6e-30f4-4f98-a20b-108233634e7b","entityId":"globalId.OIG.18.w5rdiaax2ycwhy3g2ocmpc2nhhbmqtp57ue2nycsr7bzxpilla","isPrimary":"true","entityName":"Ama Maclead","externalId":"ocid1.agcsgovernanceinstance.oc1.iad.amaaaaaaebkbezqaadpvwolr4raumlz3uxdgczwbqkalpcoo7qcu2r639054","usageName":"test create account without perm","timeCreated":1745215195989,"lastModified":1745215195989}
+{"ownershipCollectionId":"007eb58f-ba28-4dd3-b4f0-4a54dc0e8522","entityId":"globalId.OIG.18.w5rdiaax2ycwhy3g2ocmpc2nhhbmqtp57ue2nycsr7bzxpilla","isPrimary":"true","entityName":"Maclead Ama","externalId":"ocid1.agcsgovernanceinstance.oc1.iad.amaaaaaaebkbezqaadpvwolr4raumlz3uxdgczwbqkalpcoo7qcu2r639054","usageName":"CPQ_25JULY_ROLE2","timeCreated":1753425406449,"lastModified":1753425406449}
+{"ownershipCollectionId":"00a13a10-7b98-46aa-9b3f-470c639089b1","entityId":"globalId.OCI.accessgovtest_new.19bfac7c65c1007d6db122f9cbp82","isPrimary":"true","entityName":"JUNE4U3 JUNE4U3","externalId":"ocid1.agcsgovernanceinstance.oc1.iad.amaaaaaaebkbezqaadpvwolr4raumlz3uxdgczwbqkalpcoo7qcu2r639054","usageName":"AB-FAcpm","timeCreated":1749622529341,"lastModified":1749622529341}