glt-stream: deploy minimalistics stats collector (WIP)

2 files changed, 253 insertions, 0 deletions

diff --git a/files/glt/stream-stats.go b/files/glt/stream-stats.go
new file mode 100644
index 00000000..5dec4e8f
--- /dev/null
+++ b/files/glt/stream-stats.go
@@ -0,0 +1,190 @@
+package main
+
+import (
+	"crypto/sha256"
+	"encoding/json"
+	"fmt"
+	"io/ioutil"
+	"log"
+	"net/http"
+	"os"
+	"strconv"
+	"sync"
+	"time"
+)
+
+type LatestRequests map[string]bool
+
+var last5min LatestRequests
+var lMutex = &sync.Mutex{}
+
+const dateFormat = time.RFC3339
+
+func init() {
+	last5min = make(LatestRequests)
+}
+
+// find the next timestamp (i.e. time when minute is 4 mod 5 and second is 0)
+func nextTimestamp() time.Time {
+	now := time.Now()
+	if now.Minute()%5 == 4 && now.Second() == 0 {
+		return now.Add(5 * 60 * time.Second)
+	}
+
+	minDiff := 5
+	switch now.Minute() % 5 {
+	case 0:
+		minDiff = 4
+	case 1:
+		minDiff = 3
+	case 2:
+		minDiff = 2
+	case 3:
+		minDiff = 1
+	case 4:
+		minDiff = 5
+	}
+	return now.Add(-time.Duration(now.Second()) * time.Second).Add(time.Duration(minDiff) * 60 * time.Second)
+}
+
+// find the previous timestamp
+func previousTimestamp() time.Time {
+	now := time.Now()
+	if now.Minute()%5 == 4 && now.Second() == 0 {
+		return now.Add(-5 * 60 * time.Second)
+	}
+
+	minDiff := (now.Minute() % 5) + 1
+	return now.Add(-time.Duration(now.Second()) * time.Second).Add(-time.Duration(minDiff) * 60 * time.Second)
+}
+
+// writeToFile writes the 5min result to the file by appending data
+func writeToFile() {
+	filePath := os.Args[2]
+	timestamp := time.Now().Add(-5 * 60 * time.Second)
+	db := make(map[string]uint32)
+
+	// collect new count and erase data from 5-minutes data structure
+	lMutex.Lock()
+	latestCount := len(last5min)
+	last5min = make(LatestRequests)
+	lMutex.Unlock()
+
+	// read in existing data
+	content, err := ioutil.ReadFile(filePath)
+	if err == nil {
+		srcData := make(map[string]uint32)
+		err = json.Unmarshal(content, &srcData)
+		if err != nil {
+			fmt.Fprintf(os.Stderr, "failed to unmarshal file '%s': %s\n", filePath, err.Error())
+			return
+		}
+
+		// copy data over to database
+		for k, v := range srcData {
+			db[k] = v
+		}
+	}
+
+	// update database with latest count
+	db[timestamp.Format(dateFormat)] = uint32(latestCount)
+
+	// write to file
+	dump, _ := json.MarshalIndent(db, "", "  ")
+	err = ioutil.WriteFile(filePath, dump, 0644)
+	if err != nil {
+		fmt.Fprintf(os.Stderr, "error while writing file '%s': %s\n", filePath, err.Error())
+	}
+}
+
+// handle a request to /
+func handle(w http.ResponseWriter, r *http.Request) {
+	w.Header().Add("Content-type", "text/plain; charset=utf-8")
+	_, err := w.Write([]byte("request counter\nby meisterluk\nroutes: {/req, /list}\n"))
+	if err != nil {
+		fmt.Fprintln(os.Stderr, err)
+	}
+}
+
+// handle a request to /req
+// increments the counter within 5min plus one unless this client was already registered
+func handleRequest(w http.ResponseWriter, r *http.Request) {
+	// generate a key which detects trivial double requests
+	var ident string
+	ident = r.Header.Get("User-Agent")
+	if r.Header.Get("X-FORWARDED-FOR") != "" {
+		ident += r.Header.Get("X-FORWARDED-FOR")
+	}
+	if r.RemoteAddr != "" {
+		ident += r.RemoteAddr
+	}
+	//ident += time.Now().Format(dateFormat) // add this line to register every request for debugging
+	h := sha256.New()
+	key := string(h.Sum([]byte(ident)))
+
+	// register this client for counting
+	lMutex.Lock()
+	last5min[key] = true
+	defer lMutex.Unlock()
+
+	w.Write([]byte("request registered\n"))
+}
+
+func handleList(w http.ResponseWriter, r *http.Request) {
+	w.Header().Add("Content-type", "text/plain; charset=utf-8")
+	srcFile := os.Args[2]
+	db := make(map[string]uint32)
+
+	// add entry for current data
+	db[previousTimestamp().Format(dateFormat)] = uint32(len(last5min))
+
+	// read file
+	for {
+		content, err := ioutil.ReadFile(srcFile)
+		if err != nil {
+			fmt.Fprintln(os.Stderr, err)
+			break
+		}
+		srcDB := make(map[string]uint32)
+		err = json.Unmarshal(content, &srcDB)
+		if err != nil {
+			fmt.Fprintln(os.Stderr, err)
+			break
+		}
+
+		// copy data into db
+		for k, v := range srcDB {
+			db[k] = v
+		}
+		break // for loop used only for control flow
+	}
+
+	// print data
+	for timestamp, count := range db {
+		w.Write([]byte(timestamp + "\t" + strconv.Itoa(int(count)) + "\n"))
+	}
+}
+
+func main() {
+	if len(os.Args) != 3 {
+		fmt.Fprintln(os.Stderr, "usage: ./req-counter <int:port> <str:data-filepath>")
+		os.Exit(1)
+	}
+
+	http.HandleFunc("/", handle)
+	http.HandleFunc("/req", handleRequest)
+	http.HandleFunc("/list", handleList)
+
+	go func() {
+		for {
+			now := time.Now()
+			time.Sleep(nextTimestamp().Sub(now))
+
+			writeToFile()
+			fmt.Fprintf(os.Stderr, "File '%s' written.\n", os.Args[2])
+		}
+	}()
+
+	fmt.Println("listening on " + os.Args[1])
+	log.Fatal(http.ListenAndServe(os.Args[1], nil))
+}
diff --git a/spreadspace/glt-stream.yml b/spreadspace/glt-stream.yml
index 355c989f..48541d85 100644
--- a/spreadspace/glt-stream.yml
+++ b/spreadspace/glt-stream.yml
@@ -46,3 +46,66 @@
             autoindex: {}
     include_role:
       name: nginx/vhost
+
+  - name: install golang
+    apt:
+      name: go
+      state: present
+
+  - name: create base directory for stats
+    file:
+      path: /srv/www/stats
+      state: directory
+
+  - name: add user for stats
+    user:
+      name: stats
+      system: yes
+      home: /srv/www/stats
+
+  - name: create data and gocache directories for stats
+    loop:
+      - data
+      - .gocache
+    file:
+      path: "/srv/www/stats/{{ item }}"
+      state: directory
+      group: stats
+      mode: 0775
+
+  - name: install stats collector script
+    copy:
+      src: "{{ global_files_dir }}/glt/stream-stats.go"
+      dest: /srv/www/stats/stream-stats.go
+
+  - name: install systemd unit for stats collector
+    copy:
+      content: |
+        [Unit]
+        Description=GLT21 Stream Stats Collector
+
+        [Service]
+        Type=simple
+        Environment="GOCACHE=/srv/www/stats/.gocache"
+        ExecStart=/usr/bin/go run /srv/www/stats/stream-stats.go 127.0.0.1:4200 /srv/www/stats/data/glt21.json
+        NoNewPrivileges=yes
+        PrivateTmp=yes
+        PrivateDevices=yes
+        ProtectSystem=strict
+        ReadWritePaths=/srv/www/stats/data /srv/www/stats/.gocache
+        ProtectHome=yes
+        ProtectKernelTunables=yes
+        ProtectControlGroups=yes
+        RestrictRealtime=yes
+        RestrictAddressFamilies=AF_INET
+
+        [Install]
+        WantedBy=multi-user.target
+      dest: /etc/systemd/system/stream-stats.service
+
+  - name: make sure stats collector service unit is enabled and started
+    systemd:
+      name: stream-stats.service
+      daemon_reload: yes
+      enabled: yes
+      state: started