Tmux based observability tool and diffwatch json diff utility

.gitignore

@@ -89,6 +89,8 @@ docker/mongodb-kubernetes-tests/.test_identifiers*
 
 logs-debug/
 /ssdlc-report/*
+tools/mdbdebug/bin*
+tools/diffwatch/bin*
 .gocache/
 
 docs/**/log/*

controllers/operator/mongodbshardedcluster_controller.go

@@ -2868,6 +2868,10 @@ func (r *ShardedClusterReconcileHelper) statefulsetLabels() map[string]string {
 	return merge.StringToStringMap(r.sc.Labels, r.sc.GetOwnerLabels())
 }
 
+func (r *ShardedClusterReconcileHelper) DesiredShardsConfiguration() map[int]*mdbv1.ShardedClusterComponentSpec {
+	return r.desiredShardsConfiguration
+}
+
 func (r *ShardedClusterReconcileHelper) ShardsMemberClustersMap() map[int][]multicluster.MemberCluster {
 	return r.shardsMemberClustersMap
 }

tools/diffwatch/Dockerfile

@@ -0,0 +1,15 @@
+FROM alpine:latest
+
+RUN apk add --update --no-cache python3 py3-pip && ln -sf python3 /usr/bin/python
+RUN apk add bash tmux kubectl htop less fzf yq lnav
+RUN pip install tmuxp --break-system-packages
+
+# Create directory for lnav formats
+RUN mkdir -p /root/.lnav/formats/installed/
+
+COPY bin_linux/diffwatch /usr/local/bin/
+COPY lnav/*.json /root/.lnav/formats/installed/
+ADD retry_cmd.sh /usr/local/bin/
+RUN chmod +x /usr/local/bin/retry_cmd.sh
+
+CMD ["/bin/bash"]

tools/diffwatch/build.sh

@@ -0,0 +1,16 @@
+#!/usr/bin/env bash
+
+set -Eeou pipefail
+
+script_name=$(readlink -f "${BASH_SOURCE[0]}")
+script_dir=$(dirname "${script_name}")
+
+pushd "${script_dir}" >/dev/null 2>&1
+mkdir bin bin_linux >/dev/null 2>&1 || true
+
+echo "Building diffwatch from $(pwd) directory"
+GOOS=linux GOARCH=amd64 go build -o bin_linux ./...
+go build -o bin ./...
+
+echo "Copying diffwatch from $(pwd) to ${PROJECT_DIR}/bin"
+cp bin/diffwatch "${PROJECT_DIR}"/bin

tools/diffwatch/build_docker.sh

@@ -0,0 +1,8 @@
+#!/usr/bin/env bash
+
+set -Eeou pipefail
+
+export TAG=${TAG:-"latest"}
+
+docker build --platform linux/amd64 -t "quay.io/lsierant/diffwatch:${TAG}" .
+docker push "quay.io/lsierant/diffwatch:${TAG}"

tools/diffwatch/cmd/diffwatch/diffwatch.go

@@ -0,0 +1,105 @@
+package main
+
+import (
+	"context"
+	"flag"
+	"fmt"
+	"io"
+	"log"
+	"os"
+	"os/signal"
+	"path"
+	"strings"
+	"syscall"
+
+	"github.com/mongodb/mongodb-kubernetes/diffwatch/pkg/diffwatch"
+)
+
+type arrayFlags []string
+
+func (i *arrayFlags) String() string {
+	return strings.Join(*i, ",")
+}
+
+func (i *arrayFlags) Set(value string) error {
+	*i = append(*i, value)
+	return nil
+}
+
+func main() {
+	ctx, cancel := context.WithCancel(context.Background())
+
+	signalChan := make(chan os.Signal, 1)
+	signal.Notify(signalChan, syscall.SIGINT, syscall.SIGTERM)
+
+	go func() {
+		<-signalChan
+		cancel()
+	}()
+
+	readFromStdin := isInPipeMode()
+	var inputStream io.Reader
+	if readFromStdin {
+		inputStream = os.Stdin
+	}
+
+	var filePath string
+	var destDir string
+	var linesAfter int
+	var linesBefore int
+	var linesContext int
+	var ignores arrayFlags
+	flag.StringVar(&filePath, "file", "", "Path to the JSON file that will be periodically observed for changes. Optional when the content is piped on stdin. Required when -destDir is specified. "+
+		"If reading from stdin, then path is not relevant (file won't be read), but the file name will be used for the diff files prefix stored in destDir.")
+	flag.StringVar(&destDir, "destDir", "", "Path to the destination directory to store diffs. Optional. If not set, then diff files won't be created. "+
+		"If specified, then -file parameter is required. The files will be prefixed with file name of the -file parameter.")
+	flag.IntVar(&linesAfter, "A", 0, "Number of lines printed after a match (default 0)")
+	flag.IntVar(&linesBefore, "B", 0, "Number of lines printed before a match (default 0)")
+	flag.IntVar(&linesContext, "C", 3, "Number of context lines printed before and after (equivalent to setting -A and -B) (default = 3)")
+	flag.Var(&ignores, "ignore", "Regex pattern to ignore triggering diff if the only changes are ignored ones; you can specify multiple --ignore parameters, e.g. --ignore timestamp --ignore '\\d{4}-\\d{2}-\\d{2}T\\d{2}:\\d{2}:\\d{2}\\.\\d{3}Z' (ignore all lines with changed timestamp)")
+	flag.Parse()
+
+	for ignore := range ignores {
+		fmt.Println("ignore = ", ignore)
+	}
+
+	if linesBefore == 0 {
+		linesBefore = linesContext
+	}
+	if linesAfter == 0 {
+		linesAfter = linesContext
+	}
+
+	if err := watchChanges(ctx, filePath, destDir, inputStream, linesBefore, linesAfter, ignores); err != nil {
+		cancel()
+		if err == io.EOF {
+			log.Printf("Reached end of stream. Exiting.")
+		} else {
+			log.Printf("Error: %v", err)
+		}
+		os.Exit(1)
+	}
+}
+
+func isInPipeMode() bool {
+	stat, _ := os.Stdin.Stat()
+	return (stat.Mode() & os.ModeCharDevice) == 0
+}
+
+func watchChanges(ctx context.Context, filePath string, destDir string, inputStream io.Reader, linesBefore int, linesAfter int, ignores []string) error {
+	diffWriterFunc := diffwatch.WriteDiffFiles(destDir, path.Base(filePath))
+	jsonDiffer, err := diffwatch.NewJsonDiffer(linesBefore, linesAfter, diffWriterFunc, ignores)
+	if err != nil {
+		return err
+	}
+
+	// parsedFileChannel is filled in the background by reading from stream or watching the file periodically
+	parsedFileChannel := make(chan diffwatch.ParsedFileWrapper)
+	if inputStream != nil {
+		go diffwatch.ReadAndParseFromStream(ctx, inputStream, filePath, parsedFileChannel)
+	} else {
+		go diffwatch.ReadAndParseFilePeriodically(ctx, filePath, diffwatch.DefaultWatchInterval, parsedFileChannel)
+	}
+
+	return diffwatch.WatchFileChangesPeriodically(ctx, filePath, parsedFileChannel, jsonDiffer.FileChangedHandler)
+}

tools/diffwatch/cmd/diffwatch/diffwatch_int_test.go

@@ -0,0 +1,120 @@
+package main
+
+import (
+	"bytes"
+	"context"
+	"fmt"
+	"io"
+	"os"
+	"testing"
+	"time"
+
+	"github.com/mongodb/mongodb-kubernetes/diffwatch/pkg/diffwatch"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// ignored project-root/tmp dir where test outputs will be stored
+const tmpDir = "../../../../tmp/diffwatch"
+const cleanupAfterTest = false
+
+// TestDiffWatcherFromFile is a manual test that triggers simulated sequence of changes.
+// Intended for manual inspection of files.
+//
+// How to run:
+//  1. Create ops-manager-kubernetes/tmp directory
+//  2. Comment t.Skip and run the test
+//  3. View latest files:
+//     find $(find tmp/diffwatch -d 1 -type d | sort -n | tail -n 1) -type f | sort -rV | fzf --preview 'cat {}'
+func TestDiffWatcherFromFile(t *testing.T) {
+	t.Skip("Test intended to manual run, comment skip to run")
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	require.NoError(t, os.MkdirAll(tmpDir, 0770))
+	tempDir, err := os.MkdirTemp(tmpDir, time.Now().Format("20060102_150405"))
+	require.NoError(t, err)
+	defer func() {
+		if cleanupAfterTest {
+			_ = os.RemoveAll(tempDir)
+		}
+	}()
+
+	watchedFile := fmt.Sprintf("%s/watched.json", tempDir)
+	go watchChanges(ctx, watchedFile, tempDir, nil, 4, 4, []string{"ignoredField", `\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}`})
+
+	diffwatch.DefaultWatchInterval = time.Millisecond * 100
+	diffwatch.DefaultProgressInterval = time.Millisecond * 200
+	applyChange(t, "resources/base.json", watchedFile)
+	applyChange(t, "resources/changed_1.json", watchedFile)
+	applyChange(t, "resources/changed_2.json", watchedFile)
+	time.Sleep(diffwatch.DefaultProgressInterval * 2)
+	applyChange(t, "resources/changed_3.json", watchedFile)
+	applyChange(t, "resources/changed_3_ignored_only.json", watchedFile)
+	applyChange(t, "resources/changed_3_ignored_only_2.json", watchedFile)
+	applyChange(t, "resources/changed_4_ignored_only_ts.json", watchedFile)
+	applyChange(t, "resources/changed_4_ignored_only_ts_and_other.json", watchedFile)
+	applyChange(t, "resources/changed_5.json", watchedFile)
+}
+
+func TestDiffWatcherFromStream(t *testing.T) {
+	t.Skip("Test intended to manual run, comment skip to run")
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	buf := bytes.Buffer{}
+	files := []string{
+		"resources/base.json",
+		"resources/changed_1.json",
+		"resources/changed_2.json",
+		"resources/changed_3.json",
+		"resources/changed_4_ignored_only_ts.json",
+		"resources/changed_4_ignored_only_ts_and_other.json",
+		"resources/changed_5.json",
+	}
+	for _, file := range files {
+		fileBytes, err := os.ReadFile(file)
+		require.NoError(t, err)
+		buf.Write(fileBytes)
+	}
+
+	require.NoError(t, os.MkdirAll(tmpDir, 0770))
+	tempDir, err := os.MkdirTemp(tmpDir, time.Now().Format("20060102_150405"))
+	require.NoError(t, err)
+	defer func() {
+		if cleanupAfterTest {
+			_ = os.RemoveAll(tempDir)
+		}
+	}()
+
+	watchedFile := "watched.file"
+	_ = watchChanges(ctx, watchedFile, tempDir, &buf, 2, 2, []string{"ignoredField", `\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}`})
+	cancel()
+	time.Sleep(time.Second * 1)
+}
+
+func applyChange(t *testing.T, srcFilePath string, dstFilePath string) {
+	assert.NoError(t, copyFile(srcFilePath, dstFilePath, 0660))
+	time.Sleep(diffwatch.DefaultWatchInterval * 2)
+}
+
+func copyFile(srcFilePath string, dstFilePath string, mode os.FileMode) error {
+	source, err := os.Open(srcFilePath)
+	if err != nil {
+		return err
+	}
+	defer func() {
+		_ = source.Close()
+	}()
+
+	destination, err := os.OpenFile(dstFilePath, os.O_RDWR|os.O_CREATE|os.O_TRUNC, mode)
+	if err != nil {
+		return err
+	}
+	defer func() {
+		_ = destination.Close()
+	}()
+
+	_, err = io.Copy(destination, source)
+	return err
+}

tools/diffwatch/cmd/diffwatch/resources/base.json

@@ -0,0 +1,38 @@
+{
+  "version": 1,
+  "processes": [
+    {
+      "name": "om-backup-db-0-0",
+      "disabled": false,
+      "hostname": "om-backup-db-0-0-svc.lsierant-10.svc.cluster.local",
+      "args2_6": {
+        "net": {
+          "port": 27017,
+          "tls": {
+            "certificateKeyFile": "/var/lib/mongodb-automation/secrets/certs/LFKN25MS7RP2OSSJM3ORWIGPUW7VHJ24MOYDC2IXP77ADT45OR3A",
+            "mode": "requireTLS"
+          }
+        },
+        "replication": {
+          "replSetName": "om-backup-db"
+        },
+        "storage": {
+          "dbPath": "/data"
+        },
+        "systemLog": {
+          "destination": "file",
+          "logAppend": false,
+          "path": "/var/log/mongodb-mms-automation/mongodb.log"
+        }
+      },
+      "featureCompatibilityVersion": "6.0",
+      "processType": "mongod",
+      "version": "6.0.5-ent",
+      "authSchemaVersion": 5,
+      "LogRotate": {
+        "timeThresholdHrs": 0,
+        "sizeThresholdMB": 0
+      }
+    }
+  ]
+}

tools/diffwatch/cmd/diffwatch/resources/changed_1.json

@@ -0,0 +1,44 @@
+{
+  "version": 1,
+  "processes": [
+    {
+      "name": "om-backup-db-0-0",
+      "disabled": true,
+      "hostname": "om-backup-db-0-0-svc.lsierant-10.svc.cluster.local",
+      "args2_6": {
+        "net": {
+          "port": 27017,
+          "tls": {
+            "certificateKeyFile": "/var/lib/mongodb-automation/secrets/certs/LFKN25MS7RP2OSSJM3ORWIGPUW7VHJ24MOYDC2IXP77ADT45OR3A",
+            "mode": "preferTLS"
+          }
+        },
+        "replication": {
+          "replSetName": "om-backup-db"
+        },
+        "storage": {
+          "dbPath": "/data"
+        },
+        "systemLog": {
+          "destination": "file",
+          "logAppend": false,
+          "path": "/var/log/mongodb-mms-automation/mongodb.log"
+        }
+      },
+      "featureCompatibilityVersion": "6.0",
+      "processType": "mongod",
+      "version": "6.0.6-ent",
+      "authSchemaVersion": 5,
+      "newField1": 1,
+      "newField2": 2,
+      "newField3": 3,
+      "newField4": 4,
+      "newField5": 5,
+      "LogRotate": {
+        "timeThresholdHrs": 0,
+        "sizeThresholdMB": 0
+      },
+      "newField6": 6
+    }
+  ]
+}