[sources.oathkeeper_file]
  type = "file"
  include = ["/var/log/oathkeeper/access.log"]
  read_from = "beginning"

[transforms.oathkeeper_parse]
  type = "remap"
  inputs = ["oathkeeper_file"]
  source = '''
    .raw = .message
    parsed = parse_json(.message) ?? {}

    .timestamp = to_timestamp(.timestamp) ?? now()
    .request_id = parsed.request_id ?? parsed.req_id ?? ""
    request_method = get(parsed, ["request", "method"]) ?? ""
    request_path = get(parsed, ["request", "path"]) ?? ""
    request_url = get(parsed, ["request", "url"]) ?? ""
    request_host = get(parsed, ["request", "host"]) ?? ""
    request_scheme = get(parsed, ["request", "scheme"]) ?? ""
    request_query = get(parsed, ["request", "query"]) ?? ""
    .method = parsed.method ?? parsed.http_method ?? request_method ?? ""
    .path = parsed.path ?? parsed.http_path ?? request_path ?? request_url ?? ""
    response_status = get(parsed, ["response", "status"]) ?? 0
    .status = to_int(parsed.status ?? parsed.status_code ?? response_status ?? 0) ?? 0
    .latency_ms = to_int(parsed.latency_ms ?? parsed.duration_ms ?? parsed.took ?? 0) ?? 0
    identity_id = get(parsed, ["identity", "id"]) ?? ""
    .subject = parsed.subject ?? identity_id ?? ""
    .client_ip = parsed.client_ip ?? parsed.remote_ip ?? parsed.ip ?? ""
    headers = get(parsed, ["headers"]) ?? {}
    .user_agent = parsed.user_agent
    if is_null(.user_agent) { .user_agent = get(headers, ["User-Agent"]) }
    if is_null(.user_agent) { .user_agent = "" }
    .referer = get(headers, ["Referer"]) ?? ""

    .decision = parsed.decision
    if is_null(.decision) { .decision = parsed.result }
    if is_null(.decision) { .decision = "" }

    .trace_id = parsed.trace_id
    if is_null(.trace_id) { .trace_id = "" }

    .span_id = parsed.span_id
    if is_null(.span_id) { .span_id = "" }

    .rp = parsed.rp ?? ""
    .action = parsed.action ?? ""
    .target = parsed.target ?? ""
    .rule_id = parsed.rule_id ?? get(parsed, ["rule", "id"]) ?? ""
    parsed_url = {}
    if request_url != "" { parsed_url = parse_url(request_url) ?? {} }
    query_params = get(parsed_url, ["query"]) ?? {}
    .client_id = parsed.client_id ?? get(parsed, ["client", "id"]) ?? get(query_params, ["client_id"]) ?? get(query_params, ["clientId"]) ?? ""
    .parent_session_id = parsed.parent_session_id ?? get(parsed, ["extra", "parent_session_id"]) ?? ""
    .host = parsed.host ?? request_host ?? ""
    .scheme = parsed.scheme ?? request_scheme ?? ""
    .query = parsed.query ?? request_query ?? ""
    .upstream_url = parsed.upstream_url ?? get(parsed, ["upstream", "url"]) ?? ""
    .bytes_in = to_int(parsed.bytes_in ?? parsed.request_bytes ?? 0) ?? 0
    .bytes_out = to_int(parsed.bytes_out ?? parsed.response_bytes ?? 0) ?? 0
  '''

[sinks.clickhouse]
  type = "clickhouse"
  inputs = ["oathkeeper_parse"]
  endpoint = "http://ory_clickhouse:8123"
  database = "ory"
  table = "oathkeeper_access_logs"
  compression = "gzip"