fix: convert kernel monotonic times to wall time (#268)

# Description

Adjusts timestamps from eBPF that use the kernel monotonic timer to UTC
during Flow ingestion.

This is done on a best-effort basis, because it is impossible to sample
from the monotonic timer and the wall-clock at the same instant. The
difference in the time it takes to execute these instructions should be
small enough for our purposes in practice.

## Related Issue

Fixes #204 

## Checklist

- [x] I have read the [contributing
documentation](https://retina.sh/docs/contributing).
- [x] I signed and signed-off the commits (`git commit -S -s ...`). See
[this
documentation](https://docs.github.com/en/authentication/managing-commit-signature-verification/about-commit-signature-verification)
on signing commits.
- [x] I have correctly attributed the author(s) of the code.
- [ ] I have tested the changes locally.
- [x] I have followed the project's style guidelines.
- [x] I have updated the documentation, if necessary.
- [x] I have added tests, if applicable.

## Screenshots (if applicable) or Testing Completed

Please add any relevant screenshots or GIFs to showcase the changes
made.

## Additional Notes

This is effectively only implemented for Unix builds as I don't know how
Windows ktime behaves (and we're only compiling these plugins for Linux
at the moment). It can be implemented on !unix in the future as
necessary.

---

Please refer to the [CONTRIBUTING.md](../CONTRIBUTING.md) file for more
information on how to contribute to this project.

Signed-off-by: Evan Baker <rbtr@users.noreply.github.com>

internal/ktime/ktime.go

@@ -0,0 +1,9 @@
+package ktime
+
+// MonotonicOffset is the calculated skew between the kernel's
+// monotonic timer and UTC.
+//
+// If clock readings were instantaneous, this would mean that
+// MonotonicTimer - MonotonicOFfset = the UTC Boot Time, but
+// that is idealized and there will be some small errror.
+var MonotonicOffset = calculateMonotonicOffset()

internal/ktime/ktime_other.go

@@ -0,0 +1,9 @@
+//go:build !unix
+
+package ktime
+
+import "time"
+
+func calculateMonotonicOffset() time.Duration {
+	return 0 * time.Nanosecond
+}

internal/ktime/ktime_unix.go

@@ -0,0 +1,25 @@
+//go:build unix
+
+package ktime
+
+import (
+	"time"
+
+	"golang.org/x/sys/unix"
+)
+
+// calculateMonotonicOffset tries to determine the offset of the kernel's
+// monotonic clock from UTC so that measurements from eBPF using the
+// monotonic clock timestamp may be adjusted to wall-time.
+//
+// These instructions do not execute instantaneously so it will always be
+// impossible to sample both clocks at exactly the same time.
+// This means that for any single process there will be constant error in
+// the accuracy of this measurement despite the nanosecond-level precision
+// of the individual clocks.
+func calculateMonotonicOffset() time.Duration {
+	mono := &unix.Timespec{}
+	now := time.Now()
+	_ = unix.ClockGettime(unix.CLOCK_BOOTTIME, mono)
+	return time.Duration(now.UnixNano() - unix.TimespecToNsec(*mono))
+}

pkg/plugin/dropreason/_cprog/drop_reason.c

@@ -184,7 +184,7 @@ static void get_packet_from_skb(struct packet *p, struct sk_buff *skb)
     p->src_ip = iphdr.saddr;
     p->dst_ip = iphdr.daddr;
     // get current timestamp in ns
-    p->ts = bpf_ktime_get_ns();
+    p->ts = bpf_ktime_get_boot_ns();
 
     if (iphdr.protocol == IPPROTO_TCP)
     {
@@ -229,7 +229,7 @@ static void get_packet_from_sock(struct packet *p, struct sock *sk)
 	#endif
 
     // get current timestamp in ns
-    p->ts = bpf_ktime_get_ns();
+    p->ts = bpf_ktime_get_boot_ns();
     p->in_filtermap = true;
     p->src_ip = saddr;
     p->dst_ip = daddr;

pkg/plugin/dropreason/dropreason_linux.go

@@ -14,24 +14,23 @@ import (
 	"time"
 	"unsafe"
 
-	kcfg "github.com/microsoft/retina/pkg/config"
-	"github.com/microsoft/retina/pkg/utils"
-
 	"github.com/cilium/cilium/api/v1/flow"
 	hubblev1 "github.com/cilium/cilium/pkg/hubble/api/v1"
 	"github.com/cilium/ebpf"
 	"github.com/cilium/ebpf/link"
 	"github.com/cilium/ebpf/perf"
 	"github.com/cilium/ebpf/rlimit"
+	"github.com/microsoft/retina/internal/ktime"
+	kcfg "github.com/microsoft/retina/pkg/config"
 	"github.com/microsoft/retina/pkg/enricher"
 	"github.com/microsoft/retina/pkg/loader"
 	"github.com/microsoft/retina/pkg/log"
 	"github.com/microsoft/retina/pkg/metrics"
 	"github.com/microsoft/retina/pkg/plugin/api"
 	plugincommon "github.com/microsoft/retina/pkg/plugin/common"
-	"go.uber.org/zap"
-
 	_ "github.com/microsoft/retina/pkg/plugin/dropreason/_cprog" // nolint
+	"github.com/microsoft/retina/pkg/utils"
+	"go.uber.org/zap"
 )
 
 //go:generate go run github.com/cilium/ebpf/cmd/bpf2go@master -cc clang-14 -cflags "-g -O2 -Wall -D__TARGET_ARCH_${GOARCH} -Wall" -target ${GOARCH} -type metrics_map_value -type drop_reason_t -type packet kprobe ./_cprog/drop_reason.c -- -I../lib/_${GOARCH} -I../lib/common/libbpf/_src -I../filter/_cprog/
@@ -356,7 +355,7 @@ func (dr *dropReason) processRecord(ctx context.Context, id int) {
 			dropKey := (dropMetricKey)(bpfEvent.Key)
 
 			fl := utils.ToFlow(
-				int64(bpfEvent.Ts),
+				ktime.MonotonicOffset.Nanoseconds()+int64(bpfEvent.Ts),
 				utils.Int2ip(bpfEvent.SrcIp).To4(), // Precautionary To4() call.
 				utils.Int2ip(bpfEvent.DstIp).To4(), // Precautionary To4() call.
 				sourcePortShort,

pkg/plugin/packetparser/_cprog/packetparser.c

@@ -147,7 +147,7 @@ static void parse(struct __sk_buff *skb, direction d)
 	__builtin_memset(&p, 0, sizeof(p));
 
 	// Get current time in nanoseconds.
-	p.ts = bpf_ktime_get_ns();
+	p.ts = bpf_ktime_get_boot_ns();
 
 	p.dir = d;
 	p.bytes = skb->len;

pkg/plugin/packetparser/packetparser_linux.go

@@ -21,6 +21,7 @@ import (
 	"github.com/cilium/ebpf/rlimit"
 	tc "github.com/florianl/go-tc"
 	helper "github.com/florianl/go-tc/core"
+	"github.com/microsoft/retina/internal/ktime"
 	"github.com/microsoft/retina/pkg/common"
 	kcfg "github.com/microsoft/retina/pkg/config"
 	"github.com/microsoft/retina/pkg/enricher"
@@ -29,17 +30,16 @@ import (
 	"github.com/microsoft/retina/pkg/metrics"
 	"github.com/microsoft/retina/pkg/plugin/api"
 	plugincommon "github.com/microsoft/retina/pkg/plugin/common"
+	_ "github.com/microsoft/retina/pkg/plugin/lib/_amd64"             // nolint
+	_ "github.com/microsoft/retina/pkg/plugin/lib/_arm64"             // nolint
+	_ "github.com/microsoft/retina/pkg/plugin/lib/common/libbpf/_src" // nolint
+	_ "github.com/microsoft/retina/pkg/plugin/packetparser/_cprog"    // nolint
 	"github.com/microsoft/retina/pkg/pubsub"
 	"github.com/microsoft/retina/pkg/utils"
 	"github.com/microsoft/retina/pkg/watchers/endpoint"
 	"github.com/vishvananda/netlink"
 	"go.uber.org/zap"
 	"golang.org/x/sys/unix"
-
-	_ "github.com/microsoft/retina/pkg/plugin/lib/_amd64"             // nolint
-	_ "github.com/microsoft/retina/pkg/plugin/lib/_arm64"             // nolint
-	_ "github.com/microsoft/retina/pkg/plugin/lib/common/libbpf/_src" // nolint
-	_ "github.com/microsoft/retina/pkg/plugin/packetparser/_cprog"    // nolint
 )
 
 //go:generate go run github.com/cilium/ebpf/cmd/bpf2go@master -cc clang-14 -cflags "-g -O2 -Wall -D__TARGET_ARCH_${GOARCH} -Wall" -target ${GOARCH} -type packet packetparser ./_cprog/packetparser.c -- -I../lib/_${GOARCH} -I../lib/common/libbpf/_src -I../filter/_cprog/
@@ -534,7 +534,7 @@ func (p *packetParser) processRecord(ctx context.Context, id int) {
 			destinationPortShort := uint32(utils.HostToNetShort(bpfEvent.DstPort))
 
 			fl := utils.ToFlow(
-				int64(bpfEvent.Ts),
+				ktime.MonotonicOffset.Nanoseconds()+int64(bpfEvent.Ts),
 				utils.Int2ip(bpfEvent.SrcIp).To4(), // Precautionary To4() call.
 				utils.Int2ip(bpfEvent.DstIp).To4(), // Precautionary To4() call.
 				sourcePortShort,