“

Series: S6 Platform Engineering · Article 8 | Level: Advanced | Reading Time: 18 min

Audit Trail System: Complete Design for 13 Event Types

#TL;DR

The coomia-dip audit trail system captures 13 categories of critical security events: authentication, authorization decisions, data access, data modification, masking operations, classification changes, policy changes, schema changes, action execution, system configuration, export operations, anomaly detection, and compliance checks. Each event type includes a standardized event structure, contextual information, and correlation chains. This article covers the complete architecture from event classification, storage and querying, alert integration, to compliance reporting.

#1. The Need for Audit Trails

#1.1 Compliance-Driven Requirements

Modern data platforms face stringent compliance requirements where audit trails are no longer optional:

GDPR Article 30: Requires maintaining records of processing activities
SOX Act: Requires complete audit trails for financial data
HIPAA: Requires health data access logs retained for 6 years
PCI DSS: Requires logging of all access to cardholder data

#1.2 Security Operations Needs

Post-incident investigation: Trace operation chains after security events
Anomaly detection: Discover unusual access patterns from audit logs
Accountability: Determine responsible parties for data breaches
Change tracking: Record history of system configuration and policy changes

#1.3 Comparison with Palantir Foundry

Capability	Palantir Foundry	coomia-dip
Event types	Not disclosed	13 standardized event types
Event structure	Internal format	Standardized Protobuf
Storage	Internal storage	Iceberg persistence
Querying	Internal tools	gRPC API + SDK
Alert integration	Limited	OpenTelemetry integration

#2. Audit Architecture Design

#2.1 System Architecture

Code

┌─────────────────────────────────────────────────┐
│              Application Layer                   │
│    (Control Layer / Data Layer / Intelligence)    │
└──────────────┬───────────────────────────────────┘
               │ AuditEvent (Protobuf)
┌──────────────▼───────────────────────────────────┐
│             Audit Event Bus                       │
│         (Async Event Pipeline)                    │
│  ┌──────────┐  ┌──────────┐  ┌───────────────┐  │
│  │ Validator │→│ Enricher │→│ Router        │  │
│  └──────────┘  └──────────┘  └───────┬───────┘  │
└──────────────────────────────────────┼───────────┘
                 ┌─────────────────────┼──────────┐
                 │                     │          │
        ┌────────▼──────┐  ┌─────────▼───┐  ┌──▼──────────┐
        │ Iceberg Store │  │ Alert Engine │  │ SIEM Export │
        │ (Persistence) │  │ (Alerting)   │  │ (External)  │
        └───────────────┘  └─────────────┘  └─────────────┘

#2.2 Base Event Model

Python

class AuditEvent(BaseModel):
    """Base audit event model"""

    # Event identification
    event_id: str = Field(default_factory=lambda: str(uuid.uuid4()))
    event_type: AuditEventType = Field(description="Event type")
    event_subtype: str = Field(default="", description="Event subtype")

    # Temporal information
    timestamp: datetime = Field(default_factory=datetime.utcnow)
    duration_ms: int | None = Field(default=None, description="Operation duration")

    # Subject information
    subject: AuditSubject = Field(description="Operation subject")

    # Resource information
    resource: AuditResource | None = Field(default=None, description="Target resource")

    # Operation information
    action: str = Field(description="Operation action")
    outcome: AuditOutcome = Field(description="Operation result")

    # Context
    context: AuditContext = Field(description="Audit context")

    # Change details
    changes: list[AuditChange] | None = Field(default=None, description="Change details")

    # Correlation
    correlation_id: str | None = Field(default=None, description="Correlation ID")
    parent_event_id: str | None = Field(default=None, description="Parent event ID")

    # Classification
    classification_level: ClassificationLevel | None = Field(default=None)
    risk_level: RiskLevel = Field(default=RiskLevel.LOW)


class AuditSubject(BaseModel):
    """Audit subject"""
    subject_type: str          # user, service, system
    subject_id: str
    subject_name: str
    ip_address: str | None = None
    user_agent: str | None = None
    session_id: str | None = None
    roles: list[str] = Field(default_factory=list)


class AuditResource(BaseModel):
    """Audit resource"""
    resource_type: str         # object_type, dataset, action, policy
    resource_id: str
    resource_name: str
    namespace: str | None = None


class AuditOutcome(str, Enum):
    SUCCESS = "success"
    FAILURE = "failure"
    DENIED = "denied"
    ERROR = "error"
    PARTIAL = "partial"


class AuditChange(BaseModel):
    """Change record"""
    field: str
    old_value: Any | None = None
    new_value: Any | None = None
    change_type: str  # create, update, delete


class RiskLevel(str, Enum):
    LOW = "low"
    MEDIUM = "medium"
    HIGH = "high"
    CRITICAL = "critical"

#3. 13 Audit Event Types in Detail

#3.1 Event Type Enumeration

Python

class AuditEventType(str, Enum):
    """13 Audit Event Types"""

    # Identity & Access
    AUTHENTICATION = "authentication"           # 1. Authentication
    AUTHORIZATION = "authorization"             # 2. Authorization decisions

    # Data Operations
    DATA_ACCESS = "data_access"                 # 3. Data access
    DATA_MODIFICATION = "data_modification"     # 4. Data modification
    DATA_MASKING = "data_masking"               # 5. Masking operations

    # Metadata Changes
    CLASSIFICATION_CHANGE = "classification_change"  # 6. Classification changes
    POLICY_CHANGE = "policy_change"             # 7. Policy changes
    SCHEMA_CHANGE = "schema_change"             # 8. Schema changes

    # Business Operations
    ACTION_EXECUTION = "action_execution"       # 9. Action execution

    # System Administration
    SYSTEM_CONFIG = "system_config"             # 10. System configuration
    DATA_EXPORT = "data_export"                 # 11. Export operations

    # Security & Compliance
    ANOMALY_DETECTION = "anomaly_detection"     # 12. Anomaly detection
    COMPLIANCE_CHECK = "compliance_check"       # 13. Compliance checks

#3.2 Event 1: Authentication (AUTHENTICATION)

Records all authentication attempts including successes and failures:

Python

class AuthenticationEvent(AuditEvent):
    """Authentication audit event"""
    event_type: AuditEventType = AuditEventType.AUTHENTICATION

    auth_method: str           # password, oauth2, api_key, certificate
    auth_provider: str         # internal, ldap, oidc
    mfa_used: bool = False
    failure_reason: str | None = None
    login_attempt_count: int = 1

#3.3 Event 2: Authorization (AUTHORIZATION)

Records the decision process and result of every permission check:

Python

class AuthorizationEvent(AuditEvent):
    """Authorization decision audit event"""
    event_type: AuditEventType = AuditEventType.AUTHORIZATION

    permission_requested: str
    permission_granted: bool
    policy_ids: list[str]
    evaluation_layers: list[str]   # RBAC, ABAC, ReBAC
    deny_reason: str | None = None
    evaluation_time_ms: float = 0

#3.4 Event 3: Data Access (DATA_ACCESS)

Records read operations on Ontology objects:

Python

class DataAccessEvent(AuditEvent):
    """Data access audit event"""
    event_type: AuditEventType = AuditEventType.DATA_ACCESS

    query_type: str              # get, list, search, aggregate
    object_type: str
    fields_accessed: list[str]
    filter_criteria: dict | None = None
    result_count: int = 0
    classification_accessed: ClassificationLevel | None = None

#3.5 Event 4: Data Modification (DATA_MODIFICATION)

Records create, update, and delete operations on data:

Python

class DataModificationEvent(AuditEvent):
    """Data modification audit event"""
    event_type: AuditEventType = AuditEventType.DATA_MODIFICATION

    modification_type: str       # create, update, delete, bulk_update
    object_type: str
    object_id: str
    fields_modified: list[str]
    changes: list[AuditChange]
    batch_size: int = 1
    transaction_id: str | None = None

#3.6 Event 5: Data Masking (DATA_MASKING)

Records every masking operation by the dynamic masking engine:

Python

class DataMaskingEvent(AuditEvent):
    """Data masking audit event"""
    event_type: AuditEventType = AuditEventType.DATA_MASKING

    object_type: str
    fields_masked: list[FieldMaskingDetail]
    masking_policy_ids: list[str]
    query_id: str
    result_row_count: int
    masking_duration_ms: float

#3.7 Event 6: Classification Change (CLASSIFICATION_CHANGE)

Records changes to data classification levels:

Python

class ClassificationChangeEvent(AuditEvent):
    """Classification change audit event"""
    event_type: AuditEventType = AuditEventType.CLASSIFICATION_CHANGE

    object_type: str
    field_name: str
    old_classification: ClassificationLevel
    new_classification: ClassificationLevel
    change_reason: str
    approval_id: str | None = None
    is_downgrade: bool = False
    risk_level: RiskLevel = RiskLevel.MEDIUM

#3.8 Event 7: Policy Change (POLICY_CHANGE)

Records creation, modification, and deletion of security policies:

Python

class PolicyChangeEvent(AuditEvent):
    """Policy change audit event"""
    event_type: AuditEventType = AuditEventType.POLICY_CHANGE

    policy_type: str             # rbac, abac, rebac, masking
    policy_id: str
    policy_name: str
    change_type: str             # create, update, delete, enable, disable
    changes: list[AuditChange]
    effective_scope: str         # global, namespace, object_type
    risk_level: RiskLevel = RiskLevel.HIGH

#3.9 Event 8: Schema Change (SCHEMA_CHANGE)

Records changes to the Ontology Schema:

Python

class SchemaChangeEvent(AuditEvent):
    """Schema change audit event"""
    event_type: AuditEventType = AuditEventType.SCHEMA_CHANGE

    schema_type: str             # object_type, link_type, action_type
    schema_id: str
    schema_name: str
    change_type: str             # create, update, delete
    changes: list[AuditChange]
    migration_required: bool = False
    backward_compatible: bool = True

#3.10 Event 9: Action Execution (ACTION_EXECUTION)

Records Ontology Action executions:

Python

class ActionExecutionEvent(AuditEvent):
    """Action execution audit event"""
    event_type: AuditEventType = AuditEventType.ACTION_EXECUTION

    action_type: str
    action_id: str
    input_parameters: dict       # Masked input parameters
    execution_status: str        # pending, running, completed, failed
    affected_objects: list[str]
    side_effects: list[str]
    execution_time_ms: float

#3.11 Event 10: System Configuration (SYSTEM_CONFIG)

Records system configuration changes:

Python

class SystemConfigEvent(AuditEvent):
    """System configuration audit event"""
    event_type: AuditEventType = AuditEventType.SYSTEM_CONFIG

    config_scope: str            # platform, service, tenant
    config_key: str
    old_value: str | None = None
    new_value: str               # Sensitive values auto-masked
    requires_restart: bool = False
    risk_level: RiskLevel = RiskLevel.HIGH

#3.12 Event 11: Data Export (DATA_EXPORT)

Records data export operations:

Python

class DataExportEvent(AuditEvent):
    """Data export audit event"""
    event_type: AuditEventType = AuditEventType.DATA_EXPORT

    export_format: str           # csv, json, parquet, excel
    object_types: list[str]
    row_count: int
    file_size_bytes: int
    destination: str             # download, s3, sftp
    classification_levels: list[ClassificationLevel]
    export_approved: bool = True
    risk_level: RiskLevel = RiskLevel.HIGH

#3.13 Event 12: Anomaly Detection (ANOMALY_DETECTION)

Records system-detected anomalous behaviors:

Python

class AnomalyDetectionEvent(AuditEvent):
    """Anomaly detection audit event"""
    event_type: AuditEventType = AuditEventType.ANOMALY_DETECTION

    anomaly_type: str            # unusual_access, brute_force, data_exfiltration
    anomaly_score: float         # 0.0 - 1.0
    baseline_metric: str
    baseline_value: float
    observed_value: float
    detection_model: str
    related_events: list[str]    # Related event IDs
    risk_level: RiskLevel = RiskLevel.CRITICAL

#3.14 Event 13: Compliance Check (COMPLIANCE_CHECK)

Records results of automated compliance checks:

Python

class ComplianceCheckEvent(AuditEvent):
    """Compliance check audit event"""
    event_type: AuditEventType = AuditEventType.COMPLIANCE_CHECK

    regulation: str              # gdpr, hipaa, pci_dss, sox
    check_type: str              # data_retention, access_review, classification_review
    check_result: str            # pass, fail, warning
    findings: list[ComplianceFinding]
    remediation_required: bool = False
    due_date: datetime | None = None

#4. Audit Event Collector

#4.1 Async Event Collection

Python

class AuditCollector:
    """Audit event collector - async non-blocking"""

    def __init__(
        self,
        buffer_size: int = 10000,
        flush_interval: float = 5.0,
        max_batch_size: int = 500,
    ):
        self._buffer: asyncio.Queue[AuditEvent] = asyncio.Queue(maxsize=buffer_size)
        self._flush_interval = flush_interval
        self._max_batch_size = max_batch_size
        self._writers: list[AuditWriter] = []

    async def emit(self, event: AuditEvent) -> None:
        """Emit audit event (non-blocking)"""
        try:
            self._buffer.put_nowait(event)
        except asyncio.QueueFull:
            await self._emergency_flush()
            self._buffer.put_nowait(event)

    async def _flush_loop(self) -> None:
        while True:
            await asyncio.sleep(self._flush_interval)
            await self._flush()

    async def _flush(self) -> None:
        batch = []
        while len(batch) < self._max_batch_size:
            try:
                event = self._buffer.get_nowait()
                batch.append(event)
            except asyncio.QueueEmpty:
                break

        if batch:
            for writer in self._writers:
                await writer.write_batch(batch)

#4.2 Decorator-Based Integration

Python

def audit_tracked(
    event_type: AuditEventType,
    action: str,
    risk_level: RiskLevel = RiskLevel.LOW,
):
    """Audit tracking decorator"""
    def decorator(func):
        @functools.wraps(func)
        async def wrapper(*args, **kwargs):
            start_time = datetime.utcnow()
            context = get_request_context()

            try:
                result = await func(*args, **kwargs)
                outcome = AuditOutcome.SUCCESS
                return result
            except PermissionError:
                outcome = AuditOutcome.DENIED
                raise
            except Exception:
                outcome = AuditOutcome.ERROR
                raise
            finally:
                duration = (datetime.utcnow() - start_time).total_seconds() * 1000
                event = AuditEvent(
                    event_type=event_type,
                    action=action,
                    outcome=outcome,
                    subject=context.to_audit_subject(),
                    duration_ms=int(duration),
                    risk_level=risk_level,
                    context=context.to_audit_context(),
                )
                await audit_collector.emit(event)

        return wrapper
    return decorator

#5. Storage and Querying

#5.1 Iceberg Persistence

Python

class IcebergAuditWriter(AuditWriter):
    """Iceberg audit event writer"""

    TABLE_SCHEMA = Schema(
        NestedField(1, "event_id", StringType(), required=True),
        NestedField(2, "event_type", StringType(), required=True),
        NestedField(3, "timestamp", TimestampType(), required=True),
        NestedField(4, "subject_id", StringType(), required=True),
        NestedField(5, "subject_type", StringType()),
        NestedField(6, "action", StringType(), required=True),
        NestedField(7, "outcome", StringType(), required=True),
        NestedField(8, "resource_type", StringType()),
        NestedField(9, "resource_id", StringType()),
        NestedField(10, "risk_level", StringType()),
        NestedField(11, "classification_level", IntegerType()),
        NestedField(12, "details_json", StringType()),
        NestedField(13, "correlation_id", StringType()),
    )

    PARTITION_SPEC = PartitionSpec(
        PartitionField(source_id=3, field_id=1000, transform=DayTransform(), name="day"),
        PartitionField(source_id=2, field_id=1001, transform=IdentityTransform(), name="event_type"),
    )

#5.2 Query API

Python

class AuditQueryService:
    """Audit query service"""

    async def query_events(
        self, filters: AuditQueryFilters, pagination: Pagination,
    ) -> PagedResult[AuditEvent]:
        ...

    async def get_user_activity_timeline(
        self, user_id: str, start_time: datetime, end_time: datetime,
    ) -> list[AuditEvent]:
        ...

    async def get_resource_audit_trail(
        self, resource_type: str, resource_id: str,
    ) -> list[AuditEvent]:
        ...

    async def generate_compliance_report(
        self, regulation: str, period: DateRange,
    ) -> ComplianceReport:
        ...

#6. Alerting and Notification

#6.1 Alert Rule Engine

Python

class AuditAlertEngine:
    """Audit alert engine"""

    ALERT_RULES = [
        AlertRule(
            name="brute_force_detection",
            condition="event_type == 'authentication' AND outcome == 'failure'",
            threshold=5,
            window_minutes=10,
            group_by="subject.ip_address",
            severity=AlertSeverity.CRITICAL,
        ),
        AlertRule(
            name="high_classification_access",
            condition="event_type == 'data_access' AND classification_level >= 6",
            threshold=1,
            window_minutes=0,
            severity=AlertSeverity.HIGH,
        ),
        AlertRule(
            name="mass_data_export",
            condition="event_type == 'data_export' AND row_count > 100000",
            threshold=1,
            window_minutes=0,
            severity=AlertSeverity.HIGH,
        ),
        AlertRule(
            name="policy_change_outside_hours",
            condition="event_type == 'policy_change' AND NOT is_business_hours(timestamp)",
            threshold=1,
            window_minutes=0,
            severity=AlertSeverity.CRITICAL,
        ),
    ]

#7. Testing Strategy

Python

class TestAuditTrail:
    async def test_authentication_event_recorded(self):
        collector = InMemoryAuditCollector()
        service = AuthService(audit_collector=collector)

        await service.login(username="test", password="pass")

        events = collector.get_events(AuditEventType.AUTHENTICATION)
        assert len(events) == 1
        assert events[0].outcome == AuditOutcome.SUCCESS

    async def test_denied_access_recorded(self):
        collector = InMemoryAuditCollector()
        service = OntologyService(audit_collector=collector)

        with pytest.raises(PermissionError):
            await service.get_object("SecretType", "obj-1")

        events = collector.get_events(AuditEventType.AUTHORIZATION)
        assert len(events) == 1
        assert events[0].outcome == AuditOutcome.DENIED

    async def test_correlation_chain(self):
        collector = InMemoryAuditCollector()
        correlation_id = str(uuid.uuid4())

        events = collector.get_events_by_correlation(correlation_id)
        assert len(events) == 4
        assert [e.event_type for e in events] == [
            AuditEventType.AUTHENTICATION,
            AuditEventType.AUTHORIZATION,
            AuditEventType.DATA_ACCESS,
            AuditEventType.DATA_MASKING,
        ]

    async def test_buffer_overflow_handling(self):
        collector = AuditCollector(buffer_size=10)
        for i in range(20):
            await collector.emit(make_event(f"event-{i}"))

#8. Production Best Practices

#8.1 Retention Policies

Event Type	Hot Storage	Warm Storage	Cold Storage	Total Retention
Auth/Authz	30 days	180 days	3 years	3 years
Data Access	90 days	365 days	5 years	5 years
Policy/Classification	365 days	3 years	7 years	7 years
Anomaly Detection	365 days	3 years	7 years	7 years
Compliance Checks	365 days	5 years	10 years	10 years

#8.2 Performance Targets

Event collection latency: < 10ms (P99)
Buffer capacity: 10,000 events
Batch write frequency: Every 5 seconds or 500 events
Query response time: < 500ms (hot storage)

#8.3 Security Considerations

Audit logs are immutable (append-only) and cannot be modified or deleted
Access to audit logs requires independent access controls
Sensitive fields (passwords, tokens) are automatically masked before writing
The audit system's own anomalies must also be recorded (meta-audit)

#9. Summary

The coomia-dip audit trail system achieves full-chain audit coverage from authentication to compliance checks through 13 standardized event types. Key design decisions:

Comprehensive events: 13 types covering security, data, metadata, and system domains
Async non-blocking: Audit collection does not impact main business flow performance
Iceberg persistence: Supports long-term retention and efficient querying
Alert integration: Rule-based real-time alerting detects anomalous behavior
Compliance alignment: Built-in reporting capabilities for GDPR, HIPAA, SOX, and more

The next article will explore the coomia-dip dual-layer data lineage tracking system.