Skip to main content

Creating Events

This guide provides reference documentation for the audit trail event structure and creation methods. These methods are available in the API but not accessible through the Python client.

note

The Python client only supports reading events via search_events(). The methods create_event() and batch_create_events() will return 405 Method Not Allowed errors.

To maintain audit trail integrity, event creation is restricted to internal H2O AI Cloud services only.

This documentation is provided for reference only. See Searching Events for supported Python client capabilities.

Event Structure

Every audit event must include these required fields:

from datetime import datetime, timezone
from h2o_audit_trail.event.event import Event
from h2o_audit_trail.model.status import Status

event = Event(
event_time=datetime.now(timezone.utc), # When the event occurred
event_source="h2oai-my-service", # Container image name
action="actions/my-service/create", # Action identifier
read_only=False, # Whether action is read-only
status=Status(code=0), # Request status (0=OK)
principal="users/user-id", # Authenticated user
source_ip_address="192.168.1.100" # Source IP address
)

Single Event Creation

Basic event creation

The following example shows how to create a single audit event:

import uuid
from datetime import datetime, timezone

import h2o_audit_trail
from h2o_audit_trail.event.event import Event
from h2o_audit_trail.model.status import Status

# Initialize client
clients = h2o_audit_trail.login()
event_client = clients.event_client

# Create event
event_id = str(uuid.uuid4())
created_event = event_client.create_event(
event_id=event_id,
event=Event(
event_time=datetime.now(timezone.utc),
event_source="h2oai-enginemanager-server",
action="actions/enginemanager/daiEngines/CREATE",
read_only=False,
status=Status(code=0),
principal="users/john.doe",
source_ip_address="10.0.1.50"
)
)

print(f"Created event: {created_event.name}")

Event with additional metadata

You can include optional fields and metadata when creating events:

event = Event(
event_time=datetime.now(timezone.utc),
event_source="h2oai-enginemanager-server",
action="actions/enginemanager/daiEngines/CREATE",
read_only=False,
status=Status(code=0),
principal="users/john.doe",
source_ip_address="10.0.1.50",
# Optional fields
resource="//engine-manager/workspaces/my-workspace/daiEngines/my-engine",
workspace="workspaces/my-workspace",
user_agent="h2o-python-client/1.0.0",
metadata={
"engine_type": "dai",
"instance_size": "large",
"operation_id": "op-12345"
},
request_parameters={
"timeout": "3600",
"auto_start": "true"
}
)

created_event = event_client.create_event(event_id=str(uuid.uuid4()), event=event)

Batch event creation

To create multiple events efficiently, use batch creation:

from h2o_audit_trail.event.create import CreateEventRequest

# Prepare multiple events
requests = []

for i in range(10):
event = Event(
event_time=datetime.now(timezone.utc),
event_source="h2oai-batch-processor",
action=f"actions/batch/process/{i}",
read_only=True,
status=Status(code=0),
principal="services/batch-processor",
source_ip_address="private",
metadata={"batch_id": "batch-001", "item_index": str(i)}
)

requests.append(CreateEventRequest(
event_id=str(uuid.uuid4()),
event=event
))

# Submit batch (maximum 1000 events per batch)
response = event_client.batch_create_events(requests=requests)

print(f"Successfully created: {len(response.events)} events")
print(f"Failed to create: {len(response.failed_requests)} events")

# Handle partial failures
for index, rpc_status in response.failed_requests.items():
print(f"Request {index} failed with code {rpc_status.code}: {rpc_status.message}")

Event fields reference

Required fields

FieldTypeDescriptionExample
event_timedatetimeWhen the event occurred (UTC)datetime.now(timezone.utc)
event_sourcestrContainer image name following HAIC naming"h2oai-enginemanager-server"
actionstrAction identifier (should match AuthZ actions)"actions/enginemanager/daiEngines/CREATE"
read_onlyboolWhether the action is read-onlyFalse
statusStatusRequest status with code and optional detailsStatus(code=0)
principalstrAuthenticated principal identifier"users/user-id" or "services/service-name"
source_ip_addressstrSource IP (private IPs become "private")"192.168.1.100" or "private"

Optional fields

FieldTypeDescription
resourcestrTarget resource path
workspacestrRelated workspace (format: workspaces/*)
user_agentstrClient user agent
metadataDict[str, str]Service-specific key-value data
request_parametersDict[str, str]Request parameters

Status codes

Common status codes following gRPC conventions:

  • 0 - OK (success)
  • 1 - CANCELLED
  • 2 - UNKNOWN
  • 3 - INVALID_ARGUMENT
  • 4 - DEADLINE_EXCEEDED
  • 5 - NOT_FOUND
  • 6 - ALREADY_EXISTS
  • 7 - PERMISSION_DENIED
  • 13 - INTERNAL
  • 16 - UNAUTHENTICATED

Best practices

Event naming conventions

Use consistent naming patterns for better searchability:

# Good: Hierarchical action names
action="actions/enginemanager/daiEngines/CREATE"
action="actions/mlops/models/DEPLOY"
action="actions/dataset/import/VALIDATE"

# Good: Event source matching container names
event_source="h2oai-enginemanager-server"
event_source="h2oai-mlops-api"

Metadata guidelines

Keep metadata keys consistent and descriptive:

# Good: Descriptive metadata
metadata={
"operation_type": "create",
"resource_type": "dai_engine",
"workspace_id": "ws-12345",
"user_session_id": "session-abc123"
}

# Metadata key requirements:
# - 1-63 characters
# - lowercase alphanumeric or underscore
# - start with alphabetic character
# - end with alphanumeric character

Error handling

Always handle potential API errors:

from h2o_audit_trail.exception import CustomApiException

try:
event = event_client.create_event(event_id=event_id, event=event)
logger.info(f"Event created successfully: {event.name}")
except CustomApiException as e:
logger.error(f"Failed to create event: {e}")
# Handle error appropriately (retry, log, alert, etc.)

Performance considerations

  • Use batch creation for multiple events (up to 1000 per batch)
  • Consider using read-only events for queries and searches
  • Include relevant metadata for efficient searching
  • Use appropriate timestamp precision (typically second-level is sufficient)

Error scenarios

Common validation errors

# Invalid event_id (must be UUID4)
try:
event_client.create_event(event_id="invalid-id", event=event)
except CustomApiException as e:
print(f"Invalid event ID: {e}")

# Invalid metadata key
event = Event(
# ... required fields ...
metadata={"Invalid-Key": "value"} # Uppercase not allowed
)

Handling batch failures

response = event_client.batch_create_events(requests=large_batch)

# Log successful events
for event in response.events:
logger.info(f"Created event: {event.name}")

# Handle failed events
for index, rpc_status in response.failed_requests.items():
failed_request = large_batch[index]
logger.error(f"Failed to create event {failed_request.event_id}: {rpc_status.message}")

# Optionally retry individual failed events
if rpc_status.code in [2, 4, 13]: # UNKNOWN, DEADLINE_EXCEEDED, INTERNAL
# These might be transient, consider retrying
pass

Next steps


Feedback