Logging Integration - Support Tailing Logs

[maruiz93]

Goal

The goal of this feature is to refactor how the different information related to the logging backend system is stored to provide more flexibility to the KubeArchive API in the ways of retrieving logs of the containers from archived Pods, specially to allow retrieving only the last piece of information of a log.

Current Situation

Currently we store the full "log_url" in the log_url table at archival time of the Pods for each hosted container.

The problem with this is that the query params are fixed, and there is no way to provide flexible body or query params. This restricts the possibility of having more than one call like "give me the last 10 lines of the logs" or "give me the last 50 lines of the log".

Nevertheless, having a fixed url at the archival time is useful for situations like changing the backend logging system. If some logs are archived in system 1 and now the new container logs go to system 2, while both are available the logging integration will keep working smoothly without the need of a migration of all the logs. We want this to continue happening.

There is also something else we want to keep and is the extraction of some valuable information at the archival time for the container:

• START - Start timestamp
• END - End timestamp
• QUERY - The string used in that backend system to identify uniquely the logs of the container

Those are currently embedded in the fixed url, this is not needed but the information still needs to be stored in the database at archival time.

Proposed Solution

We need part of the URL to be fixed (protocol + hostname + port), the START timestamp, the QUERY to uniquely identify the container in the backend logging system, and the NAMESPACE where the container is hosted.

The rest of the information needs to be flexible: path, params, method, and JSON path.

We also need to provide all the flexible information for the different fixed urls, in case more than one backend logging system is being used and at least the information for two types of retrieval: full and tailed.

Schema

Writer

The sink will keep reading the same keys in kubearchive-logging (LOG_URL) and a couple of new ones: START, QUERY and END. If one is not provided, they will be filled with NULL.

An example of how the kubearchive-logging will look like attending to the information needed by the writer is the following:

apiVersion: v1
kind: ConfigMap
metadata:
  name: kubearchive-logging
data:
  NAMESPACE: "cel:metadata.namespace"
  POD_ID: "cel:metadata.uid"
  START: "cel:status.?startTime == optional.none() ? int(now()-duration('1h'))*1000000000: status.startTime"
  QUERY: {stream="{NAMESPACE}"} | pod_id = `{POD_ID}` | container = `{CONTAINER_NAME}`
  LOG_URL: "http://loki-gateway.grafana-loki.svc.cluster.local:80"

Note that in this example, NAMESPACE and POD_ID are added to fill what's needed for QUERY.

Reader

The kubearchive-logging ConfigMap will now include an embedded YAML with all the flexible configuration for each host to be read by the KubeArchive API.

The schema is as follows:

{
  "$schema": "http://json-schema.org/draft-07/schema#",
  "title": "Log Provider Configuration",
  "type": "object",
  "additionalProperties": {
    "$ref": "#/definitions/ProviderStrategies"
  },
  "definitions": {
    "ProviderStrategies": {
      "type": "object",
      "properties": {
        "tail": { "$ref": "#/definitions/ApiEndpoint" },
        "full": { "$ref": "#/definitions/ApiEndpoint" }
      },
      "additionalProperties": false
    },
    "ApiEndpoint": {
      "type": "object",
      "required": ["reverse", "path", "method", "json-path"],
      "properties": {
        "reverse": {
          "type": "boolean",
          "description": "Whether to reverse the output order."
        },
        "path": {
          "type": "string",
          "description": "The API endpoint path."
        },
        "method": {
          "type": "string",
          "enum": ["GET", "POST"],
          "description": "The HTTP method used for the request."
        },
        "params": {
          "type": "object",
          "description": "Query parameters for GET requests.",
          "additionalProperties": {
            "type": ["string", "number", "boolean"]
          }
        },
        "body": {
          "type": "object",
          "description": "The request body for POST requests."
        },
        "json-path": {
          "type": "string",
          "description": "JSONPath expression to extract the logs from the response."
        }
      }
    }
  }
}

Example

https://loki:
  tail:
    reverse: true
    path: /loki/api/v1/query_range
    method: GET
    params:
      query: ${QUERY}
      start: ${START}
      end: ${${START} + duration(24h)}
      limit: 50
      direction: backward
    json-path: $.data.result[*].values[*][1]
  full:
    reverse: false
    path: /loki/api/v1/query_range
    method: GET
    params:
      query: ${QUERY}
      start: ${START}
      end: ${${START} + duration(24h)}
      limit: 10000
      direction: forward
    json-path: $.data.result[*].values[*][1]
https://datadog:
  tail:
    reverse: true
    path: /api/v2/logs/events/search
    method: POST
    body:
      {
       "filter": {
       "query": "${QUERY}",
       "indexes": [
         "${NAMESPACE}"
        ],
       "from": "${START}",
       "to": "${START} + 24h"
      },
       "sort": "-timestamp",
       "page": {
         "limit": 50
       }
      }
    json-path: $.data[*].attributes.message
  full:
    reverse: false
    path: /api/v2/logs/events/search
    method: POST
    body:
      {
       "filter": {
       "query": "${QUERY}",
       "indexes": [
         "${NAMESPACE}"
        ],
       "from": "${START}",
       "to": "${START} + 24h"
      },
       "sort": "timestamp",
       "page": {
         "limit": 10000
       }
      }
    json-path: $.data[*].attributes.message
https://elasticsearch:
  tail:
    reverse: true
    path: /fluentd/_search
    method: GET
    params:
      _source_includes: message
      size: 50
      sort: _doc:desc
      q: ${QUERY}
    json-path: $.hits.hits[*]._source.message
  full:
    reverse: false
    path: /fluentd/_search
    method: GET
    params:
      _source_includes: message
      size: 10000
      sort: _doc
      q: ${QUERY}
    json-path: $.hits.hits[*]._source.message
https://splunk:
  tail:
    reverse: true
    path: /services/search/jobs/export
    method: GET
    params:
      search: search * | ${QUERY} | sort - time | head 50 | table "message"
    json-path: $[*].result.message
  full:
    reverse: false
    path: /services/search/jobs/export
    method: GET
    params:
      search: search * | ${QUERY} | table "message"
    json-path: $[*].result.message

Database Migration

Current log_url Table Schema

CREATE TABLE public.log_url (
    id BIGSERIAL PRIMARY KEY,
    uuid uuid NOT NULL REFERENCES public.resource(uuid) ON DELETE CASCADE,
    url text NOT NULL,
    container_name text NOT NULL,
    json_path text,
    created_at timestamp with time zone DEFAULT now() NOT NULL,
    updated_at timestamp with time zone DEFAULT now() NOT NULL
);

Required Changes

• Drop json_path column
• Add query, start and end columns

All the new columns contents can be extracted from the url content so we need to do the migration as follows:

BEGIN;

-- 1. Add the new columns
ALTER TABLE public.log_url
    ADD COLUMN query text,
    ADD COLUMN start text,
    ADD COLUMN end text;

-- 2. Extract data from the URL and update the rows
UPDATE public.log_url
SET
    query     = substring(url from 'query=([^&]+)'),
    start     = substring(url from 'start=([^&]+)')
WHERE url ILIKE '%loki%';

-- 3. Modify the URL to only keep the host
UPDATE public.log_url
SET url = split_part(url, '/', 1);

-- 4. Remove the old field
ALTER TABLE public.log_url DROP COLUMN json_path;

COMMIT;

Note: The problem with this migration script is that the UPDATE piece is different depending on the backend logging system used. We can cover the fully supported Loki backend logging system but for others, different UPDATE statements should be used.

Acceptance Criteria

• The migration script includes the migration to the new schema
• The log_url table includes the new fields and removes the json_path
• The KubeArchive API includes a new endpoint for retrieving tailed logs
• The kubearchive-logging ConfigMap includes YAML config file hosting the way the full logs and the tailed logs should be retrieved from the different backend logging systems used
• The logging middleware reads the kubearchive-logging config and does the queries based on that configuration