Southbound observation has TTL set according to which observation is about to expire

Possible to add safety margin to outgoing TTL

Author: zluudg

Makefile

@@ -1,32 +1,17 @@
-NAME:=observation-encoder
-
-#######################################
-# VERSION SOURCE OF TRUTH FOR PROJECT #
-#######################################
-VERSION:=0.0.0
-
-OUT:=./out
-DEFAULT_INSTALLDIR:=/usr/bin
-INSTALL:=install -p -m 0755
-COMMIT:=$$(cat COMMIT 2> /dev/null || git describe --dirty=+WiP --always 2> /dev/null)
-
-.PHONY: build install clean fmt vet coverage
+VERSION:=$(shell cat ./VERSION)
+COMMIT:=$(shell git describe --dirty=+WiP --always 2> /dev/null || echo "no-vcs")
+OUT:=$(CURDIR)/out
 
+.PHONY: all build outdir test coverage fmt vet clean ko
 
 all: build
 
 build: outdir
-	go build -v -ldflags "-X 'main.version=$(VERSION)' -X 'main.commit=$(COMMIT)'" -o $(OUT)/ ./cmd/...
+	go build -v -ldflags "-X main.commit=$(COMMIT)" -o $(OUT)/ ./cmd/...
 
 outdir:
 	-mkdir -p $(OUT)
 
-clean:
-	-rm -rf $(OUT)
-
-install:
-	test -z "$(DESTDIR)" && $(INSTALL) $(OUT)/$(PROG) $(DEFAULT_INSTALLDIR) || $(INSTALL) $(OUT)/$(PROG) $(DESTDIR)$(prefix)/bin/
-
 test:
 	go test ./...
 
@@ -40,10 +25,8 @@ fmt:
 vet:
 	go vet ./...
 
-tarball: outdir
-	@test ! -f COMMIT || (echo "Trying to make tarball from extracted tarball?" && false)
-	@test ! -z $(COMMIT) || (echo "Not tracked by git?" && false)
-	@test -z $$(git status --porcelain) || (echo "won't make tarball from dirty history" && false)
-	echo "$(COMMIT)" > $(OUT)/COMMIT
-	git archive --format=tar.gz --prefix=$(NAME)/ -o $(OUT)/$(NAME)-$(VERSION).tar.gz --add-file $(OUT)/COMMIT HEAD
-	cd $(OUT) && sha256sum -b $(NAME)-$(VERSION).tar.gz > $(NAME)-$(VERSION).sha256.txt
+clean:
+	-rm -rf $(OUT)
+
+ko:
+	ko build -L -B ./cmd/observation-encoder

README.md

@@ -13,26 +13,33 @@ sudo make install
 # Run
 
 ```
-observation-encoder -config /path/to/config/file
+observation-encoder
 ```
 
 # Sample config file
 
 ```toml
-[cert]
-interval = 100
-cert_dir = "/path/to/certs/dir"
-
-[api]
-active = true
-address = "127.0.0.1"
-port = "10001"
+debug = true
+ttl_margin = 5 # five seconds added to outgoing observations TTL
 
 [nats]
 url = "nats://127.0.0.1:4222"
-subject_prefix = "observations"
-subject_southbound = "test.subject"
-ttl = 3600
+subject_southbound = "leontest.down.tapir-pop"
+observation_subject_prefix = "leontest.observations"
+
+[[nats.buckets]]
+name = "globally_new"
+ttl = 30
+
+[[nats.buckets]]
+name = "looptest"
+ttl = 10
+
+[cert]
+active = false
+
+[api]
+active = false
 
 [libtapir]
 debug = true

VERSION

@@ -0,0 +1 @@
+0.0.0

cmd/observation-encoder/main.go

@@ -21,8 +21,9 @@ import (
 	"github.com/dnstapir/observation-encoder/internal/nats"
 )
 
-const c_ENVVAR_OVERRIDE_NATS_URL = "DNSTAPIR_NATS_URL"
+const env_DNSTAPIR_NATS_URL = "DNSTAPIR_NATS_URL"
 
+/* Rewritten if building with make */
 var commit = "BAD-BUILD"
 
 type conf struct {
@@ -84,13 +85,16 @@ func main() {
 
 	confDecoder := toml.NewDecoder(file)
 	if confDecoder == nil {
-		log.Error("Problem decoding config file '%s', exiting...", configFile)
+		log.Error("Problem creating decoder for config file '%s', exiting...", configFile)
 		os.Exit(-1)
 	}
 
 	confDecoder.DisallowUnknownFields()
-	confDecoder.Decode(&mainConf)
-	file.Close() // TODO okay to close here while also using defer above?
+	err = confDecoder.Decode(&mainConf)
+	if err != nil {
+		log.Error("Problem decoding config file '%s': %s", configFile, err)
+		os.Exit(-1)
+	}
 
 	debugFlag = debugFlag || mainConf.Debug
 
@@ -105,12 +109,13 @@ func main() {
 		})
 	if err != nil {
 		log.Error("Error creating nats log: %s", err)
+		os.Exit(-1)
 	}
 
-	envNatsUrl, overrideNatsUrl := os.LookupEnv(c_ENVVAR_OVERRIDE_NATS_URL)
+	envNatsUrl, overrideNatsUrl := os.LookupEnv(env_DNSTAPIR_NATS_URL)
 	if overrideNatsUrl {
 		mainConf.Nats.Url = envNatsUrl
-		log.Info("Overriding NATS url with environment variable '%s'", c_ENVVAR_OVERRIDE_NATS_URL)
+		log.Info("Overriding NATS url with environment variable '%s'", env_DNSTAPIR_NATS_URL)
 	}
 
 	mainConf.Nats.Log = natslog
@@ -131,6 +136,7 @@ func main() {
 		})
 	if err != nil {
 		log.Error("Error creating libtapir log: %s", err)
+		os.Exit(-1)
 	}
 
 	mainConf.Libtapir.Log = libtapirlog
@@ -151,6 +157,7 @@ func main() {
 		})
 	if err != nil {
 		log.Error("Error creating app log: %s", err)
+		os.Exit(-1)
 	}
 
 	mainConf.Log = applog
@@ -173,6 +180,7 @@ func main() {
 		})
 	if err != nil {
 		log.Error("Error creating cert log: %s", err)
+		os.Exit(-1)
 	}
 
 	mainConf.Cert.Log = certlog
@@ -193,6 +201,7 @@ func main() {
 		})
 	if err != nil {
 		log.Error("Error creating API log: %s", err)
+		os.Exit(-1)
 	}
 	mainConf.Api.Log = apilog
 	mainConf.Api.App = appHandle
@@ -214,20 +223,18 @@ func main() {
 	defer signal.Stop(sigChan)
 
 	ctx, cancel := context.WithCancel(context.Background())
-	exitCh := make(chan common.Exit)
+	exitCh := make(chan common.Exit, 10) // TODO adjustable buffer?
 
 	log.Info("Starting threads...")
 
-	go appHandle.Run(ctx, exitCh)
-
-	go apiHandle.Run(ctx, exitCh)
-	go certHandle.Run(ctx, exitCh)
+	var wg sync.WaitGroup
+	wg.Go(func() { appHandle.Run(ctx, exitCh) })
+	wg.Go(func() { apiHandle.Run(ctx, exitCh) })
+	wg.Go(func() { certHandle.Run(ctx, exitCh) })
 
 	log.Info("Threads started!")
 
 	exitLoop := false
-	var wg sync.WaitGroup
-	wg.Add(1)
 	for {
 		select {
 		case s, ok := <-sigChan:
@@ -255,45 +262,29 @@ func main() {
 		}
 		if exitLoop || (sigChan == nil && exitCh == nil) {
 			log.Info("Leaving toplevel loop")
-			wg.Done()
 			break
 		}
 	}
 
-	log.Info("Cancelling, giving threads some time to finish...")
+	log.Info("Cancelling threads")
 	cancel()
-	timeout := make(chan bool, 1)
+
+	log.Info("Waiting for threads to finish")
+
+	done := make(chan struct{})
 	go func() {
-		time.Sleep(2 * time.Second)
-		timeout <- true
+		wg.Wait()
+		close(done)
 	}()
 
-TIMEOUT_LOOP:
-	for {
-		select {
-		case exit, ok := <-exitCh:
-			if ok {
-				if exit.Err != nil {
-					log.Error("%s exited with error: '%s'", exit.ID, exit.Err)
-				} else {
-					log.Info("%s done!", exit.ID)
-				}
-			} else {
-				log.Info("exit channel was closed during shutdown")
-				exitCh = nil
-			}
-		case <-timeout:
-			log.Debug("Time's up. Proceeding with shutdown.")
-			break TIMEOUT_LOOP
-		}
-		if exitCh == nil {
-			log.Warning("exit channel closed unexpectedly")
-			break TIMEOUT_LOOP
-		}
+	const shutdownTimeout = 30 * time.Second
+	select {
+	case <-done:
+		log.Info("Threads finished")
+	case <-time.After(shutdownTimeout):
+		log.Error("Timed out waiting for threads to finish after %s", shutdownTimeout)
 	}
 
-	close(exitCh)
-	wg.Wait()
 	log.Info("Exiting...")
 	os.Exit(0)
 }

internal/api/api.go

@@ -43,6 +43,12 @@ type certHandle interface {
 
 func Create(conf Conf) (*apiHandle, error) {
 	a := new(apiHandle)
+	a.id = "api"
+	a.active = conf.Active
+
+	if !a.active {
+		return a, nil
+	}
 
 	if conf.Log == nil {
 		return nil, common.ErrBadHandle
@@ -69,7 +75,6 @@ func Create(conf Conf) (*apiHandle, error) {
 	a.certs = conf.Certs
 	a.listenInterface = net.JoinHostPort(conf.Address, conf.Port)
 	a.active = conf.Active
-	a.id = "api"
 
 	return a, nil
 }
@@ -82,7 +87,10 @@ func (a *apiHandle) Run(ctx context.Context, exitCh chan<- common.Exit) {
 
 	http.HandleFunc("/api/nats_in", a.apiHandleNatsIn)
 
-	cfg := &tls.Config{GetCertificate: a.certs.GetCertificate}
+	cfg := &tls.Config{
+		GetCertificate: a.certs.GetCertificate,
+		MinVersion:     tls.VersionTLS12,
+	}
 	srv := &http.Server{
 		TLSConfig:    cfg,
 		Addr:         a.listenInterface,