Address possible issue with skewed timestamps in system log after (re)boot with power loss (#9)

* Add tests for log with skewed timestamps

* fix: introduce workaround for aligning timestamps in boot log

See #5 for details.

* Extend tests, add comments

* loki_exporter: use mktemp instead of static dir

* loki_exporter: put curl into verbose mode for tests

* loki_exporter: clean up curl options

* loki_exporter: adjust curl headers once again

In particular:

1. Tests do not require authorization, so it has been turned off.

2. Added "Connection: close" header to all curl invocations to avoid
unnecessary log warnings in the interim proxies like nginx if HTTP
keepalives are enabled.

Author: defanator

.gitignore

@@ -6,5 +6,7 @@ __pycache__
 *.pyo
 create-test-env
 run-test-exporter-onetime
+run-test-exporter-timeshifted-onetime
 tests/*.log.state
+tests/default-timeshifted.log
 loki-exporter/

Makefile

@@ -142,6 +142,9 @@ run-test-exporter-boot: test-env ## Run exporter with mocking logread (BOOT=1)
 	LOKI_AUTH_HEADER="none" \
 	/bin/bash -u loki_exporter.sh
 
+tests/default-timeshifted.log: tests/default.log
+	$(TOPDIR)/tests/create_timeshifted_log.py >$@
+
 run-test-exporter-onetime: create-test-env ## Run one-time cycle of mocking logread + exporter (BOOT=1)
 	LOGREAD="./tests/logread.py" \
 	BOOT=1 \
@@ -153,17 +156,31 @@ run-test-exporter-onetime: create-test-env ## Run one-time cycle of mocking logr
 	/bin/bash -u loki_exporter.sh
 	touch $@
 
+run-test-exporter-timeshifted-onetime: tests/default-timeshifted.log create-test-env ## Run one-time cycle of mocking logread + exporter (BOOT=1 with time unsync emulation)
+	LOGREAD="./tests/logread.py --log-file tests/default-timeshifted.log" \
+	BOOT=1 \
+	HOSTNAME="$(shell hostname).timeshifted" \
+	LOKI_PUSH_URL="http://127.0.0.1:3100/loki/api/v1/push" \
+	LOKI_AUTH_HEADER="none" \
+	MAX_FOLLOW_CYCLES=3 \
+	AUTOTEST=1 \
+	/bin/bash -u loki_exporter.sh
+	touch $@
+
 .PHONY: test
-test: run-test-exporter-onetime | .venv results ## Run tests
+test: run-test-exporter-onetime run-test-exporter-timeshifted-onetime | .venv results ## Run tests
 	$(TOPDIR)/.venv/bin/python3 -m pytest
 
 .PHONY: compare-logs
 compare-logs: | results
 	cat tests/default.log | cut -c 43- | sort >results/messages.original
 	cat results/resurrected.log | cut -c 17- | sort >results/messages.resurrected
-	wc -l results/messages.original results/messages.resurrected
-	diff -u results/messages.original results/messages.resurrected ||:
-	test $$(diff -u results/messages.original results/messages.resurrected | grep -- " (MOCK)$$" | wc -l) -eq 3
+	cat results/resurrected-timeshifted.log | cut -c 17- | sort >results/messages.resurrected-timeshifted
+	for target in messages.resurrected messages.resurrected-timeshifted; do \
+		wc -l results/messages.original results/$${target} ; \
+		diff -u results/messages.original results/$${target} ||: ; \
+		test $$(diff -u results/messages.original results/$${target} | grep -- " (MOCK)$$" | wc -l) -eq 3 ; \
+	done
 
 .PHONY: save-logs
 save-logs: | results
@@ -261,7 +278,8 @@ clean: delete-test-env ## Clean-up
 	find $(TOPDIR)/ -type f -name "*.pyo" -delete
 	find $(TOPDIR)/ -type d -name "__pycache__" -delete
 	rm -rf $(TOPDIR)/.pytest_cache
-	rm -f $(TOPDIR)/run-test-exporter-onetime
+	rm -f $(TOPDIR)/run-test-exporter-onetime $(TOPDIR)/run-test-exporter-timeshifted-onetime
 	find $(TOPDIR)/tests/ -type f -name "*.log.state" -delete
+	rm -f $(TOPDIR)/tests/default-timeshifted.log
 	rm -rf $(TOPDIR)/results
 	rm -rf $(TOPDIR)/loki-exporter

loki_exporter.sh

@@ -4,16 +4,28 @@
 # ^^^ the above line is purely for shellcheck to treat this as a bash-like script
 # (OpenWRT's ash from busybox is kinda similar but there still could be issues)
 
-PIPE_NAME="/tmp/loki_exporter.pipe"
-BULK_DATA="/tmp/loki_exporter.boot"
+_TMPDIR=$(mktemp -d -p /tmp loki_exporter.XXXXXX)
+PIPE_NAME="/${_TMPDIR}/loki_exporter.pipe"
+BULK_DATA="/${_TMPDIR}/loki_exporter.boot"
 
 LOKI_MSG_TEMPLATE="{\"streams\": [{\"stream\": {\"job\": \"openwrt_loki_exporter\", \"host\": \"${HOSTNAME}\"}, \"values\": [[\"TIMESTAMP\", \"MESSAGE\"]]}]}"
 LOKI_BULK_TEMPLATE_HEADER="{\"streams\": [{\"stream\": {\"job\": \"openwrt_loki_exporter\", \"host\": \"${HOSTNAME}\"}, \"values\": ["
 LOKI_BULK_TEMPLATE_MSG="[\"TIMESTAMP\", \"MESSAGE\"],"
 LOKI_BULK_TEMPLATE_FOOTER="]}]}"
 
+DATETIME_STR_FORMAT="%a %b %d %H:%M:%S %Y"
+OS=$(uname -s | tr "[:upper:]" "[:lower:]")
+
+if [ "${AUTOTEST-0}" -eq 1 ]; then
+    _CURL_BULK_CMD=(curl --no-progress-meter -fv -H "Content-Type: application/json" -H "Content-Encoding: gzip" -H "Connection: close")
+    _CURL_CMD=(curl --no-progress-meter -fv -H "Content-Type: application/json" -H "Connection: close")
+else
+    _CURL_BULK_CMD=(curl -fsS -H "Content-Type: application/json" -H "Content-Encoding: gzip" -H "Authorization: Basic ${LOKI_AUTH_HEADER}" -H "Connection: close")
+    _CURL_CMD=(curl -fsS -H "Content-Type: application/json" -H "Authorization: Basic ${LOKI_AUTH_HEADER}" -H "Connection: close")
+fi
+
 _setup() {
-    mkfifo ${PIPE_NAME}
+    mkfifo "${PIPE_NAME}"
     echo "started with BOOT=${BOOT}" >&2
 }
 
@@ -23,11 +35,17 @@ _teardown() {
     else
         echo "tailer ${tailer_pid} killed, exiting" >&2
     fi
-    rm -f ${PIPE_NAME}
+    rm -f "${PIPE_NAME}"
+    if [ "${AUTOTEST-0}" -eq 1 ]; then
+        mkdir -p results
+        cp -r "${_TMPDIR}" results/
+    fi
+    rm -rf "${_TMPDIR}"
     exit 0
 }
 
 _do_bulk_post() {
+    _log_file="$1"
     post_body="${LOKI_BULK_TEMPLATE_HEADER}"
 
     while read -r line; do
@@ -49,21 +67,109 @@ _do_bulk_post() {
         msg_payload="${msg_payload/MESSAGE/$msg}"
 
         post_body="${post_body}${msg_payload}"
-    done <${BULK_DATA}
+    done <"${_log_file}"
 
     post_body="${post_body:0:${#post_body}-1}${LOKI_BULK_TEMPLATE_FOOTER}"
-    echo "${post_body}" | gzip >${BULK_DATA}.payload.gz
-    rm -f ${BULK_DATA}
+    echo "${post_body}" | gzip >"${_log_file}.payload.gz"
+    rm -f "${_log_file}"
 
-    if curl -fs -X POST -H "Content-Type: application/json" -H "Content-Encoding: gzip" -H "Authorization: Basic ${LOKI_AUTH_HEADER}" --data-binary "@${BULK_DATA}.payload.gz" "${LOKI_PUSH_URL}"; then
-        if [ "${AUTOTEST-0}" -eq 1 ]; then
-            mkdir -p results
-            cp ${BULK_DATA}.payload.gz results/
+    if ! "${_CURL_BULK_CMD[@]}" --data-binary "@${_log_file}.payload.gz" "${LOKI_PUSH_URL}" >"${_log_file}.payload.gz-response" 2>&1; then
+        echo "BULK POST FAILED: leaving ${_log_file}.payload.gz for now"
+    fi
+}
+
+_check_for_skewed_timestamp() {
+    _log_file="$1"
+
+    # maximum threshold for comparing timestamps between 2 subsequent log lines (ns)
+    delta_threshold=86400000000000
+
+    # incremental step for substituting timestamps of unsynchronized log lines (ns)
+    step=25000000
+
+    # step 1: search for possible skewed timestamp
+    prev_ts=0
+    line_n=0
+    line_n_synced=0
+    while read -r line; do
+        ts="${line:26:14}"
+        ts_ms="${ts/./}"
+
+        # shellcheck disable=SC2116
+        # subshell is required to handle multiplication errors and keep the loop
+        if ! ts_ns="$(echo $(( ts_ms * 1000 * 1000 )) )" ; then
+            continue
         fi
-        rm -f ${BULK_DATA}.payload.gz
-    else
-        echo "BULK POST FAILED: leaving ${BULK_DATA}.payload.gz for now"
+
+        line_n=$((line_n + 1))
+
+        if [ "${prev_ts-0}" -eq 0 ]; then
+            prev_ts=$ts_ns
+        fi
+
+        delta_t=$((ts_ns - prev_ts))
+        if [ $delta_t -ge $delta_threshold ]; then
+            # found a line with timestamp delta exceeding a given threshold
+            line_n_synced=$line_n
+            break
+        fi
+
+        prev_ts=$ts_ns
+    done <"${_log_file}"
+
+    if [ $line_n_synced -eq 0 ]; then
+        # no skew detected, nothing to do
+        return
     fi
+
+    # skew detected, 1st synced line is $line_n_synced;
+    # round new ts to nearest second
+    ts_ns=$((ts_ns / 1000000000))
+    ts_ns=$((ts_ns * 1000000000))
+    new_ts=$((ts_ns - step * (line_n_synced-1)))
+
+    # step 2: re-create boot log with fake timestamps in appropriate range
+    rm -f "${_log_file}.new"
+    line_n=0
+    while read -r line; do
+        ts="${line:26:14}"
+        ts_ms="${ts/./}"
+
+        # shellcheck disable=SC2116
+        # subshell is required to handle multiplication errors and keep the loop
+        if ! ts_ns="$(echo $(( ts_ms * 1000 * 1000 )) )" ; then
+            continue
+        fi
+
+        line_n=$((line_n + 1))
+
+        # for lines with valid timestamps, just print a line as is
+        if [ $line_n -ge $line_n_synced ]; then
+            printf "%s\n" "$line" >>"$1.new"
+            continue
+        fi
+
+        # otherwise, craft a new line
+        msg="${line:42:2000}"
+
+        new_ts_s=$((new_ts / 1000000000))
+        case "${OS}" in
+            darwin)
+                datetime_str=$(date -r "${new_ts_s}" +"${DATETIME_STR_FORMAT}")
+                ;;
+            *)
+                datetime_str=$(date -d @"${new_ts_s}" +"${DATETIME_STR_FORMAT}")
+                ;;
+        esac
+
+        new_ts_ms_rounded=$((new_ts / 1000000))
+        printf "%s [%s] %s\n" "${datetime_str}" "${new_ts_ms_rounded:0:10}.${new_ts_ms_rounded:10:13}" "${msg}" >>"${_log_file}.new"
+
+        # increase timestamp
+        new_ts=$((new_ts + step))
+    done <"${_log_file}"
+
+    mv "${_log_file}.new" "${_log_file}"
 }
 
 _main_loop() {
@@ -78,7 +184,7 @@ _main_loop() {
         fi
     fi
 
-    ${TAILER_CMD} >${PIPE_NAME} 2>&1 &
+    ${TAILER_CMD} >"${PIPE_NAME}" 2>&1 &
     tailer_pid=$!
 
     while read -r line; do
@@ -105,21 +211,21 @@ _main_loop() {
         post_body="${post_body/TIMESTAMP/$ts_ns}"
         post_body="${post_body/MESSAGE/$msg}"
 
-        if ! curl -fs -X POST -H "Content-type: application/json" -H "Authorization: Basic ${LOKI_AUTH_HEADER}" -d "${post_body}" "${LOKI_PUSH_URL}"; then
+        if ! "${_CURL_CMD[@]}" -d "${post_body}" "${LOKI_PUSH_URL}"; then
             echo "POST FAILED: '${post_body}'"
         fi
 
         MIN_TIMESTAMP="${ts_ns}"
-    done <${PIPE_NAME}
+    done <"${PIPE_NAME}"
 }
 
 _setup
 
 MIN_TIMESTAMP=0
 
 if [ "${BOOT}" -eq 1 ]; then
-    ${LOGREAD} -t >${BULK_DATA}
-    last_line="$(tail -1 ${BULK_DATA})"
+    ${LOGREAD} -t >"${BULK_DATA}"
+    last_line="$(tail -1 "${BULK_DATA}")"
     ts="${last_line:26:14}"
     ts_ms="${ts/./}"
 
@@ -131,7 +237,8 @@ if [ "${BOOT}" -eq 1 ]; then
         MIN_TIMESTAMP=${ts_ns}
     fi
 
-    _do_bulk_post
+    _check_for_skewed_timestamp "${BULK_DATA}"
+    _do_bulk_post "${BULK_DATA}"
 fi
 
 trap "_teardown" SIGINT SIGTERM EXIT

tests/create_timeshifted_log.py

@@ -0,0 +1,41 @@
+#!/usr/bin/env python3
+"""
+Helper script to create log with timestamp skew emulation
+"""
+
+import sys
+import random
+from logread import MockLogRead
+
+
+def main():
+    """
+    Entrypoint
+    """
+    mlr = MockLogRead()
+    mlr.parse_args()
+    mlr.load_lines()
+
+    num_lines = len(mlr.lines)
+    sync_from_line = random.randrange(
+        (num_lines // 3) * 2, (num_lines // 3) * 2 + (num_lines // 3) // 2
+    )
+    shift_back = (180 * 86400) + (random.randrange(1, 30) * 86400)
+
+    # print(f"loaded {num_lines} lines; making {sync_from_line} the 1st one with synced time")
+
+    for nl, line in enumerate(mlr.lines):
+        if nl < sync_from_line:
+            msg = mlr.get_msg_from_line(line)
+            ts = mlr.get_ts_existing(line)
+            ts -= shift_back
+            tstr = mlr.datetime_str_from_ts(ts)
+            print(f"{tstr} [{ts:.03f}] {msg}")
+            continue
+        print(line)
+
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

tests/logread.py

@@ -21,6 +21,7 @@ def __init__(self):
         self.lines = []
         self.args = None
         self.ts_first_line = None
+        self.ts_last_line = None
         self.ts_start_from = None
         self.ts_apply_delta = None
 
@@ -142,6 +143,7 @@ def load_lines(self):
             self.lines = [line.rstrip() for line in file.readlines()]
 
         self.ts_first_line = self.get_adjusted_ts(self.lines[0])
+        self.ts_last_line = self.get_adjusted_ts(self.lines[-1])
         state_file = self.args.log_file + ".state"
 
         if os.path.isfile(state_file):
@@ -152,7 +154,8 @@ def load_lines(self):
             with open(state_file, "w", encoding="utf-8") as file:
                 file.write(f"{self.ts_start_from.timestamp()}\n")
 
-        self.ts_apply_delta = self.ts_start_from.timestamp() - self.ts_first_line
+        # for timeshift tests we assume that last line always contains synchronized timestamp
+        self.ts_apply_delta = self.ts_start_from.timestamp() - self.ts_last_line
 
     def print_line(self, line: str):
         """