improved daemon and service lifecycle (control)

.circleci/config.yml

@@ -579,7 +579,7 @@ jobs:
                 -e IMAGE_VERSION="${IMAGE_VERSION}" \
                 -e UNIT_VERSION="${UNIT_VERSION}" \
                 -e UNIT_ARCH=amd64 \
-                -e MESSAGES_PUSHED=100000 \
+                -e MESSAGES_PUSHED=10000000 \
                 -e NO_TTY=1 \
                 -v /sys/fs/cgroup:/sys/fs/cgroup:ro \
                 -v /var/run/docker.sock:/var/run/docker.sock \

dev/lifecycle/test

@@ -53,7 +53,7 @@ cd ${GOPATH}/src/github.com/jancajthaml-openbank/${TARGET_PACKAGE}
   -v ./... \
   -coverprofile=${coverage_out} \
   -coverpkg=./... \
-  -timeout=20s | tee ${test_out}
+  -timeout=10s | tee ${test_out}
 
 # remove anscii colors from file
 sed -i 's/\x1b\[[0-9;]*m//g' ${test_out}

packaging/debian_amd64/DEBIAN/control

@@ -1,5 +1,5 @@
 Package: lake
-Version: 1.2.2+packaging-improvements
+Version: 1.2.2+improve-service-lifecycle
 Section: misc
 Priority: extra
 Architecture: amd64

perf/messaging/publisher.py

@@ -1,5 +1,6 @@
 #!/usr/bin/env python3
 
+import time
 import zmq
 import itertools
 
@@ -26,17 +27,21 @@ def Publisher(number_of_messages):
 
   number_of_messages = int(number_of_messages)
 
-  for _ in itertools.repeat(None, number_of_messages+1):
-    try:
-      push.send(msg, zmq.NOBLOCK)
-    except:
-      pass
+  for _ in itertools.repeat(None, number_of_messages):
+    for _ in itertools.repeat(None, 1000):
+      try:
+        push.send(msg, zmq.NOBLOCK)
+        break
+      except:
+        pass
 
-  for _ in itertools.repeat(None, number_of_messages+1):
+  for _ in itertools.repeat(None, number_of_messages):
     try:
-      sub.recv(zmq.NOBLOCK)
+      sub.recv(zmq.BLOCK)
     except:
-      pass
+      break
+
+  time.sleep(2)
 
   push.disconnect(push_url)
   sub.disconnect(sub_url)

perf/metrics/aggregator.py

@@ -13,6 +13,7 @@ def __init__(self, path):
     self._stop_event = threading.Event()
     self.__store = dict()
     self.__path = path
+    self.__last_value = None
 
   def stop(self) -> None:
     self._stop_event.set()
@@ -28,7 +29,11 @@ def __process_change(self) -> None:
         data = json.load(fd)
         (i, e, m) = data['messageIngress'], data['messageEgress'], data['memoryAllocated']
         del data
-        self.__store[str(int(time.time()*1000))] = '{}/{}/{}'.format(i, e, m)
+        value = '{}/{}/{}'.format(i, e, m)
+        if value != self.__last_value:
+          self.__store[str(int(time.time()*1000))] = value
+          self.__last_value = value
+
     except:
       pass
 

perf/metrics/fascade.py

@@ -20,7 +20,7 @@ def __init__(self, filename):
     if self.max_ts is None or self.min_ts is None:
       self.duration = 0
     else:
-      self.duration = self.max_ts-self.min_ts+1
+      self.duration = self.max_ts - self.min_ts + 1
 
   def __compute_ts_boundaries(self, dataset):
     if not dataset:
@@ -42,8 +42,12 @@ def __normalise_fps(self, dataset):
     keys = list(dataset.keys())
     values = list(dataset.values())
 
-    ingress = [(keys[i], int(y.split('/')[0]) - int(x.split('/')[0])) for i, (x,y) in enumerate(zip(values, values[1:]))]
-    egress = [(keys[i], int(y.split('/')[1]) - int(x.split('/')[1])) for i, (x,y) in enumerate(zip(values, values[1:]))]
+    if len(values) == 1:
+      ingress = [(keys[0], int(values[0].split('/')[0]))]
+      egress = [(keys[0], int(values[0].split('/')[1]))]
+    else:
+      ingress = [(keys[i], int(y.split('/')[0]) - int(x.split('/')[0])) for i, (x,y) in enumerate(zip(values, values[1:]))]
+      egress = [(keys[i], int(y.split('/')[1]) - int(x.split('/')[1])) for i, (x,y) in enumerate(zip(values, values[1:]))]
 
     ingress = dict(ingress + [(keys[-1], ingress[-1][1])])
     egress = dict(egress + [(keys[-1], egress[-1][1])])

services/lake/boot/init.go

@@ -1,4 +1,4 @@
-// Copyright (c) 2016-2019, Jan Cajthaml <jan.cajthaml@gmail.com>
+// Copyright (c) 2016-2020, Jan Cajthaml <jan.cajthaml@gmail.com>
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.

services/lake/boot/run.go

@@ -1,4 +1,4 @@
-// Copyright (c) 2016-2019, Jan Cajthaml <jan.cajthaml@gmail.com>
+// Copyright (c) 2016-2020, Jan Cajthaml <jan.cajthaml@gmail.com>
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.

services/lake/config/config.go

@@ -1,4 +1,4 @@
-// Copyright (c) 2016-2019, Jan Cajthaml <jan.cajthaml@gmail.com>
+// Copyright (c) 2016-2020, Jan Cajthaml <jan.cajthaml@gmail.com>
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.

services/lake/config/environment.go

@@ -1,4 +1,4 @@
-// Copyright (c) 2016-2019, Jan Cajthaml <jan.cajthaml@gmail.com>
+// Copyright (c) 2016-2020, Jan Cajthaml <jan.cajthaml@gmail.com>
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.

services/lake/main.go

@@ -1,4 +1,4 @@
-// Copyright (c) 2016-2019, Jan Cajthaml <jan.cajthaml@gmail.com>
+// Copyright (c) 2016-2020, Jan Cajthaml <jan.cajthaml@gmail.com>
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.

services/lake/metrics/metrics.go

@@ -1,4 +1,4 @@
-// Copyright (c) 2016-2019, Jan Cajthaml <jan.cajthaml@gmail.com>
+// Copyright (c) 2016-2020, Jan Cajthaml <jan.cajthaml@gmail.com>
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
@@ -15,7 +15,6 @@
 package metrics
 
 import (
-	"fmt"
 	"sync/atomic"
 	"time"
 
@@ -32,37 +31,8 @@ func (metrics *Metrics) MessageIngress() {
 	atomic.AddUint64(metrics.messageIngress, 1)
 }
 
-// WaitReady wait for metrics to be ready
-func (metrics Metrics) WaitReady(deadline time.Duration) (err error) {
-	defer func() {
-		if e := recover(); e != nil {
-			switch x := e.(type) {
-			case string:
-				err = fmt.Errorf(x)
-			case error:
-				err = x
-			default:
-				err = fmt.Errorf("unknown panic")
-			}
-		}
-	}()
-
-	ticker := time.NewTicker(deadline)
-	select {
-	case <-metrics.IsReady:
-		ticker.Stop()
-		err = nil
-		return
-	case <-ticker.C:
-		err = fmt.Errorf("daemon was not ready within %v seconds", deadline)
-		return
-	}
-}
-
 // Start handles everything needed to start metrics daemon
 func (metrics Metrics) Start() {
-	defer metrics.MarkDone()
-
 	ticker := time.NewTicker(metrics.refreshRate)
 	defer ticker.Stop()
 
@@ -79,20 +49,25 @@ func (metrics Metrics) Start() {
 	case <-metrics.CanStart:
 		break
 	case <-metrics.Done():
+		metrics.MarkDone()
 		return
 	}
 
 	log.Infof("Start metrics daemon, update each %v into %v", metrics.refreshRate, metrics.output)
 
-	for {
-		select {
-		case <-metrics.Done():
-			log.Info("Stopping metrics daemon")
-			metrics.Persist()
-			log.Info("Stop metrics daemon")
-			return
-		case <-ticker.C:
-			metrics.Persist()
+	go func() {
+		for {
+			select {
+			case <-metrics.Done():
+				metrics.Persist()
+				metrics.MarkDone()
+				return
+			case <-ticker.C:
+				metrics.Persist()
+			}
 		}
-	}
+	}()
+
+	<-metrics.IsDone
+	log.Info("Stop metrics daemon")
 }

services/lake/metrics/model.go

@@ -1,4 +1,4 @@
-// Copyright (c) 2016-2019, Jan Cajthaml <jan.cajthaml@gmail.com>
+// Copyright (c) 2016-2020, Jan Cajthaml <jan.cajthaml@gmail.com>
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
@@ -42,7 +42,7 @@ func NewMetrics(ctx context.Context, continuous bool, output string, refreshRate
 	ingress := uint64(0)
 
 	return Metrics{
-		DaemonSupport:  utils.NewDaemonSupport(ctx),
+		DaemonSupport:  utils.NewDaemonSupport(ctx, "metrics"),
 		output:         output,
 		continuous:     continuous,
 		refreshRate:    refreshRate,

services/lake/metrics/persistence.go

@@ -1,4 +1,4 @@
-// Copyright (c) 2016-2019, Jan Cajthaml <jan.cajthaml@gmail.com>
+// Copyright (c) 2016-2020, Jan Cajthaml <jan.cajthaml@gmail.com>
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.