Initial version of daemon mode

.gitignore

@@ -1,7 +1,9 @@
+/3rdparty
 /bin
 /coverage.html
 /dist/
 /logstash-filter-verifier
 /.vscode/
 *.coverprofile
 *.swp
+logstash-filter-verifier.yml

Makefile

@@ -35,11 +35,14 @@ TARGETS := darwin_amd64 linux_386 linux_amd64 windows_386 windows_amd64
 
 VERSION := $(shell git describe --tags --always)
 
-GOCOV         := $(GOBIN)/gocov$(EXEC_SUFFIX)
-GOCOV_HTML    := $(GOBIN)/gocov-html$(EXEC_SUFFIX)
-GOLANGCI_LINT := $(GOBIN)/golangci-lint$(EXEC_SUFFIX)
-GOVVV         := $(GOBIN)/govvv$(EXEC_SUFFIX)
-OVERALLS      := $(GOBIN)/overalls$(EXEC_SUFFIX)
+GOCOV              := $(GOBIN)/gocov$(EXEC_SUFFIX)
+GOCOV_HTML         := $(GOBIN)/gocov-html$(EXEC_SUFFIX)
+GOLANGCI_LINT      := $(GOBIN)/golangci-lint$(EXEC_SUFFIX)
+GOVVV              := $(GOBIN)/govvv$(EXEC_SUFFIX)
+OVERALLS           := $(GOBIN)/overalls$(EXEC_SUFFIX)
+PROTOC_GEN_GO      := $(GOBIN)/protoc-gen-go$(EXEC_SUFFIX)
+PROTOC_GEN_GO_GRPC := $(GOBIN)/protoc-gen-go-grpc$(EXEC_SUFFIX)
+MOQ                := $(GOBIN)/moq$(EXEC_SUFFIX)
 
 GOLANGCI_LINT_VERSION := v1.32.2
 
@@ -66,11 +69,24 @@ $(GOVVV):
 $(OVERALLS):
 	go get github.com/go-playground/overalls
 
+$(PROTOC_GEN_GO):
+	go get google.golang.org/protobuf/cmd/protoc-gen-go
+
+$(PROTOC_GEN_GO_GRPC):
+	go get google.golang.org/grpc/cmd/protoc-gen-go-grpc
+
+$(MOQ):
+	go get github.com/matryer/moq
+
 # The Go compiler is fast and pretty good about figuring out what to
 # build so we don't try to to outsmart it.
-$(PROGRAM)$(EXEC_SUFFIX): .FORCE $(GOVVV)
+$(PROGRAM)$(EXEC_SUFFIX): gogenerate .FORCE $(GOVVV)
 	govvv build -o $@
 
+.PHONY: gogenerate
+gogenerate: $(MOQ)
+	go generate ./...
+
 .PHONY: check
 check: $(GOLANGCI_LINT)
 	golangci-lint run

README.md

@@ -466,6 +466,18 @@ are a couple of known quirks that are easy to work around:
   tool to use.
 
 
+## Development
+
+### Dependencies
+
+For a fully working development environment, the following tooling needs to be
+present:
+
+* Go compiler
+* `make` command
+* Proto buffer compiler (`protobuf-compiler`)
+
+
 ## Known limitations and future work
 
 * Some log formats don't include all timestamp components. For

go.mod

@@ -6,18 +6,31 @@ require (
 	github.com/Masterminds/semver/v3 v3.0.1
 	github.com/ahmetb/govvv v0.3.0
 	github.com/axw/gocov v1.0.0
-	github.com/breml/logstash-config v0.1.0
+	github.com/bmatcuk/doublestar/v2 v2.0.4
+	github.com/breml/logstash-config v0.4.0
 	github.com/go-playground/overalls v0.0.0-20191218162659-7df9f728c018
+	github.com/golang/protobuf v1.4.2
 	github.com/hashicorp/packer v1.4.4
+	github.com/hpcloud/tail v1.0.0
 	github.com/imkira/go-observer v1.0.3
 	github.com/matm/gocov-html v0.0.0-20200509184451-71874e2e203b
+	github.com/matoous/go-nanoid v1.5.0
+	github.com/matryer/is v1.4.0
+	github.com/matryer/moq v0.2.1
 	github.com/mattn/go-shellwords v1.0.6
 	github.com/mikefarah/yaml/v2 v2.4.0
 	github.com/op/go-logging v0.0.0-20160315200505-970db520ece7
+	github.com/pkg/errors v0.9.1
 	github.com/spf13/cobra v1.1.1
 	github.com/spf13/viper v1.7.1
-	github.com/stretchr/testify v1.4.0
+	github.com/stretchr/testify v1.5.1
+	github.com/tidwall/gjson v1.6.8
+	github.com/tidwall/sjson v1.1.5
 	github.com/yookoala/realpath v1.0.0 // indirect
 	golang.org/x/sys v0.0.0-20210124154548-22da62e12c0c // indirect
+	google.golang.org/grpc v1.34.0
+	google.golang.org/grpc/cmd/protoc-gen-go-grpc v1.0.1
+	google.golang.org/protobuf v1.25.0
 	gopkg.in/go-playground/assert.v1 v1.2.1 // indirect
+	gopkg.in/yaml.v2 v2.2.8
 )

integration_test.go

@@ -0,0 +1,114 @@
+package main_test
+
+import (
+	"context"
+	"os"
+	"path"
+	"testing"
+	"time"
+
+	"github.com/matryer/is"
+
+	"github.com/magnusbaeck/logstash-filter-verifier/v2/internal/app/daemon"
+	"github.com/magnusbaeck/logstash-filter-verifier/v2/internal/app/daemon/run"
+	"github.com/magnusbaeck/logstash-filter-verifier/v2/internal/daemon/api/grpc"
+	"github.com/magnusbaeck/logstash-filter-verifier/v2/internal/daemon/file"
+	"github.com/magnusbaeck/logstash-filter-verifier/v2/internal/logging"
+)
+
+func TestIntegration(t *testing.T) {
+	if os.Getenv("INTEGRATION_TEST") != "1" {
+		t.Skip("integration test skipped, enable with env var `INTEGRATION_TEST=1`")
+	}
+
+	is := is.New(t)
+
+	testLogger := &logging.LoggerMock{
+		DebugFunc:    func(args ...interface{}) { t.Log(args...) },
+		DebugfFunc:   func(format string, args ...interface{}) { t.Logf(format, args...) },
+		ErrorFunc:    func(args ...interface{}) { t.Log(args...) },
+		ErrorfFunc:   func(format string, args ...interface{}) { t.Logf(format, args...) },
+		FatalFunc:    func(args ...interface{}) { t.Log(args...) },
+		FatalfFunc:   func(format string, args ...interface{}) { t.Logf(format, args...) },
+		InfoFunc:     func(args ...interface{}) { t.Log(args...) },
+		InfofFunc:    func(format string, args ...interface{}) { t.Logf(format, args...) },
+		WarningFunc:  func(args ...interface{}) { t.Log(args...) },
+		WarningfFunc: func(format string, args ...interface{}) { t.Logf(format, args...) },
+	}
+
+	tempdir := t.TempDir()
+	// Start Daemon
+	socket := path.Join(tempdir, "integration_test.socket")
+	logstashPath := path.Join("3rdparty/logstash-7.10.0/bin/logstash")
+	if !file.Exists(logstashPath) {
+		t.Fatalf("Logstash needs to be present in %q for the integration tests to work", logstashPath)
+	}
+
+	log := testLogger
+	server := daemon.New(socket, logstashPath, log, 10*time.Second, 3*time.Second)
+
+	ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
+	defer cancel()
+
+	go func() {
+		defer cancel()
+
+		is := is.New(t)
+
+		defer server.Cleanup()
+
+		err := server.Run(ctx)
+		is.NoErr(err)
+	}()
+
+	i := 0
+	for {
+		if file.Exists(socket) {
+			break
+		}
+		time.Sleep(100 * time.Millisecond)
+		i++
+		if i >= 20 {
+			t.Fatalf("wait for socket file failed")
+		}
+	}
+
+	// Run tests
+	cases := []struct {
+		name      string
+		pipeline  string
+		basePath  string
+		testcases string
+	}{
+		{
+			name: "basic_pipeline",
+		},
+		{
+			name: "conditional_output",
+		},
+		{
+			name: "pipeline_to_pipeline",
+		},
+	}
+
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			client, err := run.New(
+				path.Join(tempdir, "integration_test.socket"),
+				log,
+				"testdata/"+tc.name+".yml",
+				"testdata/"+tc.name,
+				"testdata/testcases/"+tc.name,
+			)
+			is.NoErr(err)
+
+			err = client.Run()
+			is.NoErr(err)
+		})
+	}
+
+	_, err := server.Shutdown(context.Background(), &grpc.ShutdownRequest{})
+	is.NoErr(err)
+
+	<-ctx.Done()
+}

internal/app/app.go

@@ -78,6 +78,7 @@ func makeRootCmd(version string) *cobra.Command {
 	_ = viper.BindPFlag("loglevel", rootCmd.PersistentFlags().Lookup("loglevel"))
 
 	rootCmd.AddCommand(makeStandaloneCmd())
+	rootCmd.AddCommand(makeDaemonCmd())
 
 	return rootCmd
 }

internal/app/daemon.go

@@ -0,0 +1,22 @@
+package app
+
+import (
+	"github.com/spf13/cobra"
+	"github.com/spf13/viper"
+)
+
+func makeDaemonCmd() *cobra.Command {
+	cmd := &cobra.Command{
+		Use:   "daemon",
+		Short: "Control logstash-filter-verifier daemon mode",
+	}
+
+	cmd.PersistentFlags().StringP("socket", "s", "", "location of the control socket")
+	_ = viper.BindPFlag("socket", cmd.PersistentFlags().Lookup("socket"))
+
+	cmd.AddCommand(makeDaemonStartCmd())
+	cmd.AddCommand(makeDaemonShutdownCmd())
+	cmd.AddCommand(makeDaemonRunCmd())
+
+	return cmd
+}