You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flagon.apache.org by au...@apache.org on 2026/10/11 15:53:35 UTC
(flagon) branch fix/distill-userale-schema created (now 3507ce3)
This is an automated email from the ASF dual-hosted git repository.
brucearctor pushed a change to branch fix/distill-userale-schema
in repository https://gitbox.apache.org/repos/asf/flagon.git
at 3507ce3 fix(distill): support UserALE v2.4/v2.5 fields and normalize Proto-JSON v1 and OTLP/JSON payloads
This branch includes the following new commits:
new 3507ce3 fix(distill): support UserALE v2.4/v2.5 fields and normalize Proto-JSON v1 and OTLP/JSON payloads
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(flagon) 01/01: fix(distill): support UserALE v2.4/v2.5 fields and normalize Proto-JSON v1 and OTLP/JSON payloads
Posted by au...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
brucearctor pushed a commit to branch fix/distill-userale-schema
in repository https://gitbox.apache.org/repos/asf/flagon.git
commit 3507ce3d6ad6d34e9c0043b51ac08e4bd98470e2
Author: Bruce Arctor <50...@users.noreply.github.com>
AuthorDate: Sun Oct 11 08:52:59 2026 -0700
fix(distill): support UserALE v2.4/v2.5 fields and normalize Proto-JSON v1 and OTLP/JSON payloads
---
products/distill/distill/schemas/userale.py | 304 +++++++++++++++++++++++++---
products/distill/tests/test_log.py | 150 +++++++++++++-
2 files changed, 425 insertions(+), 29 deletions(-)
diff --git a/products/distill/distill/schemas/userale.py b/products/distill/distill/schemas/userale.py
index cc6ca02..50f3a29 100644
--- a/products/distill/distill/schemas/userale.py
+++ b/products/distill/distill/schemas/userale.py
@@ -14,21 +14,30 @@
# See the License for the specific language governing permissions and
# limitations under the License.
from datetime import datetime
-from typing import List, Optional
+from typing import Any, Dict, List, Optional, Union
-from pydantic import BaseModel, Field, AliasGenerator, field_serializer, field_validator
+from pydantic import (
+ AliasGenerator,
+ BaseModel,
+ Field,
+ field_serializer,
+ field_validator,
+ model_validator,
+)
from pydantic.alias_generators import to_camel
from pydantic.config import ConfigDict
from distill.schemas.base import BaseSchema
+
class Browser(BaseModel):
- browser: str
- version: str
+ browser: Optional[str] = None
+ version: Optional[str] = None
+
class Location(BaseModel):
- x: Optional[int]
- y: Optional[int]
+ x: Optional[Union[int, float]] = None
+ y: Optional[Union[int, float]] = None
class ScrnRes(BaseModel):
@@ -37,29 +46,268 @@ class ScrnRes(BaseModel):
class Details(BaseModel):
- window: bool
+ model_config = ConfigDict(extra="allow")
+ window: Optional[bool] = None
+
+
+def _otlp_any_value_to_python(val: Any) -> Any:
+ if not isinstance(val, dict):
+ return val
+ if "stringValue" in val:
+ return val["stringValue"]
+ if "boolValue" in val:
+ return val["boolValue"]
+ if "intValue" in val:
+ try:
+ return int(val["intValue"])
+ except (ValueError, TypeError):
+ return val["intValue"]
+ if "doubleValue" in val:
+ return val["doubleValue"]
+ if "arrayValue" in val and isinstance(val["arrayValue"], dict):
+ return [
+ _otlp_any_value_to_python(item)
+ for item in val["arrayValue"].get("values", [])
+ ]
+ if "kvlistValue" in val and isinstance(val["kvlistValue"], dict):
+ return {
+ entry["key"]: _otlp_any_value_to_python(entry.get("value"))
+ for entry in val["kvlistValue"].get("values", [])
+ if isinstance(entry, dict) and "key" in entry
+ }
+ return val
+
+
+def normalize_userale_dict(
+ data: Any, resource_attrs: Optional[Dict[str, Any]] = None
+) -> Any:
+ """
+ Normalizes a Protobuf v1 JSON UserALELog or an OpenTelemetry OTLP/JSON
+ LogRecord into the flat UserALE dictionary structure expected by Distill.
+ """
+ if not isinstance(data, dict):
+ return data
+
+ # 1. Protobuf v1 JSON (apache.flagon.userale.v1.UserALELog)
+ log_type_enum = data.get("logType")
+ if log_type_enum in (
+ "LOG_TYPE_RAW",
+ "LOG_TYPE_INTERVAL",
+ "LOG_TYPE_CUSTOM",
+ ) or any(k in data for k in ("raw", "interval", "custom")):
+ case_key = (
+ "interval"
+ if "interval" in data
+ else "custom"
+ if "custom" in data
+ else "raw"
+ )
+ inner = data.get(case_key)
+ if isinstance(inner, dict):
+ meta = inner.get("metadata") or {}
+ flat: Dict[str, Any] = {
+ "target": inner.get("target") or "#document",
+ "path": inner.get("path") or [],
+ "pageUrl": meta.get("pageUrl", ""),
+ "pageTitle": meta.get("pageTitle", ""),
+ "pageReferrer": meta.get("pageReferrer", ""),
+ "browser": {"browser": None, "version": None},
+ "userAgent": meta.get("userAgent"),
+ "type": inner.get("eventType") or case_key,
+ "logType": case_key,
+ "userAction": bool(inner.get("userAction", False)),
+ "userId": meta.get("userId"),
+ "toolVersion": meta.get("toolVersion"),
+ "toolName": meta.get("toolName"),
+ "useraleVersion": meta.get("useraleVersion"),
+ "sessionId": meta.get("sessionId"),
+ "httpSessionId": meta.get("httpSessionId"),
+ "browserSessionId": meta.get("browserSessionId"),
+ }
+ if case_key == "interval":
+ flat.update(
+ {
+ "count": int(inner.get("count", 0)),
+ "duration": int(inner.get("duration", 0)),
+ "startTime": int(inner.get("startTime", 0)),
+ "endTime": int(inner.get("endTime", 0)),
+ "targetChange": bool(inner.get("targetChange", False)),
+ "typeChange": bool(inner.get("typeChange", False)),
+ }
+ )
+ else:
+ scrn = meta.get("screenResolution") or {
+ "width": 0,
+ "height": 0,
+ }
+ loc = inner.get("location") or {"x": None, "y": None}
+ flat.update(
+ {
+ "clientTime": int(meta.get("clientTime", 0)),
+ "microTime": inner.get("microTime", 0),
+ "location": {
+ "x": loc.get("x"),
+ "y": loc.get("y"),
+ },
+ "scrnRes": {
+ "width": int(scrn.get("width", 0)),
+ "height": int(scrn.get("height", 0)),
+ },
+ "details": inner.get("details"),
+ }
+ )
+ if "attributes" in inner:
+ flat["attributes"] = inner["attributes"]
+ if "style" in inner:
+ flat["style"] = inner["style"]
+ if isinstance(inner.get("customFields"), dict):
+ flat.update(inner["customFields"])
+ return flat
+
+ # 2. OpenTelemetry OTLP/JSON LogRecord
+ if "timeUnixNano" in data and "attributes" in data and isinstance(
+ data["attributes"], list
+ ):
+ res = resource_attrs or {}
+ attrs = {
+ item["key"]: _otlp_any_value_to_python(item.get("value"))
+ for item in data["attributes"]
+ if isinstance(item, dict) and "key" in item
+ }
+ log_type = str(attrs.get("flagon.log_type", "raw"))
+ event_type = str(attrs.get("flagon.event_type", log_type))
+ time_ms = int(data["timeUnixNano"]) // 1_000_000
+ body_val = _otlp_any_value_to_python(data.get("body"))
+ details = body_val if isinstance(body_val, dict) else None
+
+ flat = {
+ "target": attrs.get("flagon.target") or "#document",
+ "path": attrs.get("flagon.path") or [],
+ "pageUrl": attrs.get("url.full", ""),
+ "pageTitle": attrs.get("page.title", ""),
+ "pageReferrer": attrs.get("page.referrer", ""),
+ "browser": {
+ "browser": res.get("browser.name"),
+ "version": res.get("browser.version"),
+ },
+ "userAgent": res.get("user_agent.original"),
+ "type": event_type,
+ "logType": log_type,
+ "userAction": bool(attrs.get("flagon.user_action", True)),
+ "userId": attrs.get("user.id"),
+ "toolVersion": res.get("service.version"),
+ "toolName": res.get("service.name"),
+ "useraleVersion": res.get("telemetry.sdk.version"),
+ "sessionId": res.get("session.id"),
+ "httpSessionId": res.get("flagon.http_session_id"),
+ "browserSessionId": res.get("flagon.browser_session_id"),
+ }
+ if data.get("traceId"):
+ flat["traceId"] = data["traceId"]
+ if data.get("spanId"):
+ flat["spanId"] = data["spanId"]
+
+ if log_type == "interval":
+ flat.update(
+ {
+ "count": int(attrs.get("flagon.interval.count", 1)),
+ "duration": int(
+ attrs.get("flagon.interval.duration_ms", 0)
+ ),
+ "startTime": int(
+ attrs.get("flagon.interval.start_time_ms", time_ms)
+ ),
+ "endTime": int(
+ attrs.get("flagon.interval.end_time_ms", time_ms)
+ ),
+ "targetChange": bool(
+ attrs.get("flagon.interval.target_change", False)
+ ),
+ "typeChange": bool(
+ attrs.get("flagon.interval.type_change", False)
+ ),
+ }
+ )
+ else:
+ flat.update(
+ {
+ "clientTime": time_ms,
+ "microTime": (int(data["timeUnixNano"]) % 1_000_000)
+ / 1_000_000,
+ "location": {
+ "x": attrs.get("flagon.location.x"),
+ "y": attrs.get("flagon.location.y"),
+ },
+ "scrnRes": {
+ "width": int(res.get("screen.width", 0)),
+ "height": int(res.get("screen.height", 0)),
+ },
+ "details": details,
+ }
+ )
+ return flat
+
+ return data
+
+
+def extract_logs(payload: Union[Dict[str, Any], List[Dict[str, Any]]]) -> List[Dict[str, Any]]:
+ """
+ Extracts a list of normalized flat UserALE log dicts from:
+ - a standard UserALE JSON array,
+ - a Protobuf v1 LogBatch dict ({"logs": [...]}), or
+ - an OTLP/JSON ExportLogsServiceRequest dict ({"resourceLogs": [...]}).
+ """
+ if isinstance(payload, list):
+ return [normalize_userale_dict(item) for item in payload]
+ if isinstance(payload, dict):
+ if "logs" in payload and isinstance(payload["logs"], list):
+ return [normalize_userale_dict(item) for item in payload["logs"]]
+ if "resourceLogs" in payload and isinstance(payload["resourceLogs"], list):
+ results: List[Dict[str, Any]] = []
+ for rl in payload["resourceLogs"]:
+ res_attrs = {
+ item["key"]: _otlp_any_value_to_python(item.get("value"))
+ for item in rl.get("resource", {}).get("attributes", [])
+ if isinstance(item, dict) and "key" in item
+ }
+ for sl in rl.get("scopeLogs", []):
+ for lr in sl.get("logRecords", []):
+ results.append(normalize_userale_dict(lr, res_attrs))
+ return results
+ return [normalize_userale_dict(payload)]
+ return []
+
class UserAleBaseSchema(BaseSchema):
"""
- A raw or custom log produced by UserAle
+ A raw, interval, or custom log produced by UserAle
"""
- target: str
- path: List[str]
+ model_config = ConfigDict(extra="allow")
+
+ target: Optional[str] = None
+ path: Optional[List[str]] = None
page_url: str
- page_title: str
- page_referrer: str
- browser: Browser
- type_field: str = Field(..., validation_alias="type", serialization_alias="type")
+ page_title: Optional[str] = None
+ page_referrer: Optional[str] = None
+ browser: Optional[Browser] = None
+ type_field: Optional[str] = Field(
+ default=None, validation_alias="type", serialization_alias="type"
+ )
log_type: str
user_action: bool
- user_id: str
- tool_version: Optional[str]
- tool_name: Optional[str]
- userale_version: Optional[str]
- session_id: str
- http_session_id: str
- browser_session_id: str
+ user_id: Optional[str] = None
+ tool_version: Optional[str] = None
+ tool_name: Optional[str] = None
+ userale_version: Optional[str] = None
+ session_id: Optional[str] = None
+ http_session_id: Optional[str] = None
+ browser_session_id: Optional[str] = None
+
+ @model_validator(mode="before")
+ @classmethod
+ def _normalize_wire_formats(cls, data: Any) -> Any:
+ return normalize_userale_dict(data)
def _timestamp(self):
"""
@@ -71,11 +319,12 @@ class UserAleBaseSchema(BaseSchema):
class UserAleIntervalSchema(UserAleBaseSchema):
"""
- A raw or custom log produced by UserAle
+ An interval log produced by UserAle
"""
model_config = ConfigDict(
title="IntervalLog",
+ extra="allow",
alias_generator=AliasGenerator(
validation_alias=to_camel, serialization_alias=to_camel
),
@@ -96,8 +345,6 @@ class UserAleIntervalSchema(UserAleBaseSchema):
def serialize_st(self, st: datetime):
return int(st.timestamp() * 1000)
- # add in end_time validator and serializer under same tag
-
def _timestamp(self):
"""
Returns:
@@ -113,16 +360,17 @@ class UserAleRawSchema(UserAleBaseSchema):
model_config = ConfigDict(
title="RawLog",
+ extra="allow",
alias_generator=AliasGenerator(
validation_alias=to_camel, serialization_alias=to_camel
),
)
client_time: int
- micro_time: int = Field(..., lt=2)
- location: Location
- scrn_res: ScrnRes
- details: Details
+ micro_time: Optional[Union[int, float]] = None
+ location: Optional[Location] = None
+ scrn_res: Optional[ScrnRes] = None
+ details: Optional[Union[Dict[str, Any], str]] = None
@field_validator("client_time")
def validate_ct(cls, ct: float):
diff --git a/products/distill/tests/test_log.py b/products/distill/tests/test_log.py
index 7b0a4a4..78f803f 100644
--- a/products/distill/tests/test_log.py
+++ b/products/distill/tests/test_log.py
@@ -21,6 +21,7 @@ from datetime import datetime
from pydantic import ValidationError
from distill.core.log import Log
+from distill.schemas.userale import extract_logs
from tests.data_config import DATA_DIR
@@ -91,6 +92,7 @@ def test_log_normalize_timestamp():
assert test_log.data.client_time == correct_dt
assert test_log.to_dict()["clientTime"] == correct_ms
+
def test_log_comparison():
data = load_log()
test_log = Log(data=data)
@@ -104,7 +106,152 @@ def test_log_comparison():
assert test_log == test_log_equal
assert test_log < test_other < datetime.now()
-
+
+
+def test_log_v2_5_and_wire_formats():
+ # 1. Modern UserALE v2.4/v2.5 raw log with float microTime, click details, userAgent, attributes, style
+ v25_raw = {
+ "target": "button#buy",
+ "path": ["button#buy", "body"],
+ "pageUrl": "https://shop.example.com",
+ "pageTitle": "Shop",
+ "pageReferrer": "",
+ "browser": {"browser": "chrome", "version": "125.0"},
+ "userAgent": "Mozilla/5.0 Chrome/125.0",
+ "clientTime": 1719530111079,
+ "microTime": 145.67,
+ "location": {"x": 100, "y": 200},
+ "scrnRes": {"width": 1920, "height": 1080},
+ "type": "click",
+ "logType": "raw",
+ "userAction": True,
+ "details": {"button": 0, "ctrl": False},
+ "userId": None,
+ "toolVersion": "1.0.0",
+ "toolName": "shop",
+ "useraleVersion": "2.5.0",
+ "sessionId": "session_1",
+ "httpSessionId": "http_1",
+ "browserSessionId": None,
+ "attributes": {"id": "buy"},
+ "style": {"display": "block"},
+ }
+ parsed_v25 = Log(data=v25_raw)
+ assert parsed_v25.data.micro_time == 145.67
+ assert parsed_v25.data.details == {"button": 0, "ctrl": False}
+ assert parsed_v25.to_dict()["userAgent"] == "Mozilla/5.0 Chrome/125.0"
+
+ # 2. Protobuf v1 JSON LogBatch
+ proto_batch = {
+ "logs": [
+ {
+ "logType": "LOG_TYPE_RAW",
+ "raw": {
+ "metadata": {
+ "pageUrl": "https://shop.example.com",
+ "pageTitle": "Shop",
+ "pageReferrer": "",
+ "userAgent": "Mozilla/5.0",
+ "clientTime": "1719530111079",
+ "screenResolution": {"width": 1920, "height": 1080},
+ "sessionId": "sess-1",
+ },
+ "target": "a#link",
+ "path": ["a#link", "body"],
+ "eventType": "click",
+ "location": {"x": 10, "y": 20},
+ "microTime": 12.5,
+ "userAction": True,
+ "details": {"button": 0},
+ },
+ }
+ ]
+ }
+ extracted_proto = extract_logs(proto_batch)
+ assert len(extracted_proto) == 1
+ proto_log = Log(data=extracted_proto[0])
+ assert proto_log.data.target == "a#link"
+ assert proto_log.data.page_url == "https://shop.example.com"
+
+ # 3. OpenTelemetry OTLP/JSON ExportLogsServiceRequest
+ otlp_payload = {
+ "resourceLogs": [
+ {
+ "resource": {
+ "attributes": [
+ {
+ "key": "service.name",
+ "value": {"stringValue": "shop"},
+ },
+ {
+ "key": "session.id",
+ "value": {"stringValue": "sess-otlp"},
+ },
+ {"key": "screen.width", "value": {"intValue": "1440"}},
+ {"key": "screen.height", "value": {"intValue": "900"}},
+ ]
+ },
+ "scopeLogs": [
+ {
+ "scope": {
+ "name": "org.apache.flagon.userale",
+ "version": "2.5.0",
+ },
+ "logRecords": [
+ {
+ "timeUnixNano": "1719530111079000000",
+ "observedTimeUnixNano": "1719530111079000000",
+ "severityNumber": 9,
+ "severityText": "INFO",
+ "eventName": "userale.raw.click",
+ "body": {
+ "kvlistValue": {
+ "values": [
+ {
+ "key": "button",
+ "value": {"intValue": "0"},
+ }
+ ]
+ }
+ },
+ "attributes": [
+ {
+ "key": "flagon.log_type",
+ "value": {"stringValue": "raw"},
+ },
+ {
+ "key": "flagon.event_type",
+ "value": {"stringValue": "click"},
+ },
+ {
+ "key": "flagon.target",
+ "value": {"stringValue": "button#pay"},
+ },
+ {
+ "key": "url.full",
+ "value": {
+ "stringValue": "https://shop.example.com/pay"
+ },
+ },
+ ],
+ "traceId": "4bf92f3577b34da6a3ce929d0e0e4736",
+ "spanId": "00f067aa0ba902b7",
+ }
+ ],
+ }
+ ],
+ }
+ ]
+ }
+ extracted_otlp = extract_logs(otlp_payload)
+ assert len(extracted_otlp) == 1
+ otlp_log = Log(data=extracted_otlp[0])
+ assert otlp_log.data.target == "button#pay"
+ assert otlp_log.data.page_url == "https://shop.example.com/pay"
+ assert otlp_log.data.details == {"button": 0}
+ assert otlp_log.to_dict()["traceId"] == "4bf92f3577b34da6a3ce929d0e0e4736"
+
+
def load_log() -> str:
with open(os.path.join(DATA_DIR, "log_test_data.json")) as f:
data = f.readline()
@@ -115,3 +262,4 @@ def load_interval_log() -> str:
with open(os.path.join(DATA_DIR, "log_interval_data.json")) as f:
data = f.readline()
return data
+