Can one log to loki with a time in the past?

[DGRFL]

I have an agent that sometimes detects elements that need to be logged "into the past".

python-logging-loki reuses the python-logging mechanism, so I tried adding a timestamp filter as specifief in the stackoverflow message (see link in source code).

Can one 'tweak' the datetime on which a log line is sent to loki?

import logging
import logging_loki
import time

# https://stackoverflow.com/questions/47146514/python-logging-using-custom-timestamps-not-current-timestamp

class TimestampFilter (logging.Filter):
        """
        This is a logging filter which will check for a `timestamp` attribute on a
        given LogRecord, and if present it will override the LogRecord creation time
        to be that of the timestamp (specified as a time.time()-style value).
        This allows one to override the date/time output for log entries by specifying
        `timestamp` in the `extra` option to the logging call.
        """

        def filter(self, record):
            if hasattr(record, 'timestamp'):
                record.created = record.timestamp
            return True

# --- Standard initialization
handler = logging_loki.LokiHandler(
    url="http://git-lapp202.gtos.psa:3100/loki/api/v1/push",
    tags={"application": "my-app"},
    version="1",
)
logger = logging.getLogger("my-logger")
logger.addHandler(handler)

# --- Add timestampfilter ...
filter = TimestampFilter()
logger.addFilter(filter)

# --- Now log a record without time
logger.error( "Logging at actual time", extra={"tags": {"service": "my-service"}},)

# --- Now log a record 15 minutes ago
my_timestamp = time.time() -900
logger.error( "Logging in the past", extra={"tags": {"service": "my-service", 'timestamp' : my_timestamp}},)

[DGRFL]

For who might be interested in pushing information in the past to loki.

I managed to alter the emitter.py code to achieve this.

class LokiEmitterV1(LokiEmitter):
    """Emitter for Loki >= 0.4.0."""

    def build_payload(self, record: logging.LogRecord, line) -> dict:
        """Build JSON payload with a log entry."""
        labels = self.build_tags(record)
        if ("loki_ts" in labels.keys()):
            lokits=(int(labels["loki_ts"]))
            ns = 1e9
            prd = int(lokits * 1e9)
            ts = str(prd)
            labels.pop("loki_ts")
        else:
            ns = 1e9
            ts = str(int(time.time() * ns))

        stream = {
            "stream": labels,
            "values": [[ts, line]],
        }
        return {"streams": [stream]}

If a "loki_ts" label is present, this is taken as timestamp, (and removed from the labels.

In order to call this, the client becomes :

print("\n\n# --- Now log a record without time")
logger.error( "Logging at actual time", extra={"tags": {"service": "my-service"}},)

print("\n\n# --- Now log a record 15 minutes ago")
my_timestamp = time.time()
my_timestamp = my_timestamp -900
logger.error( "Logging in the past", extra={"tags": {"service": "my-service", 'loki_ts' : my_timestamp}},)

It is not up to me to commit such change, but it seems to work. I still need to evaluate whether loki or grafana would "alert" on arrival of such old messages