feat(jobreceiver): Implement monitoring job receiver (#1272)

* [pkg/reciever/jobreceiver] Implement event output handler

Signed-off-by: Christian Kruse <[email protected]>

* [pkg/reciever/jobreceiver] Implement logentries output handler

Signed-off-by: Christian Kruse <[email protected]>

* [pkg/reciever/jobreceiver] Implement receiver

Implements a feature gated monitoringjob receiver that schedules
and executes commands.

Signed-off-by: Christian Kruse <[email protected]>

* remove featuregate

Signed-off-by: Christian Kruse <[email protected]>

* [pkg/reciever/jobreceiver] Implement Assets

Signed-off-by: Christian Kruse <[email protected]>

* race in test

Signed-off-by: Christian Kruse <[email protected]>

* Clean up asset fetching logic and add basic tests

Signed-off-by: Christian Kruse <[email protected]>

* spelling

Signed-off-by: Christian Kruse <[email protected]>

* prevent command from being scheduled when assets cannot be fetched

Signed-off-by: Christian Kruse <[email protected]>

* add test case for fetching invalid url

Signed-off-by: Christian Kruse <[email protected]>

---------

Signed-off-by: Christian Kruse <[email protected]>

pkg/receiver/jobreceiver/asset/config.go

@@ -1,13 +1,21 @@
 package asset
 
-import "errors"
+import (
+	"errors"
+	"fmt"
+	"net/url"
+)
+
+const (
+	binDir     = "bin"
+	libDir     = "lib"
+	includeDir = "include"
+)
 
 // Spec for asset fetching
 type Spec struct {
 	// Name is the name of the asset
 	Name string `mapstructure:"name"`
-	// Path is the absolute path to where the asset should be installed
-	Path string `mapstructure:"path"`
 	// Url is the remote address used for fetching the asset
 	URL string `mapstructure:"url"`
 	// SHA512 is the hash of the asset tarball
@@ -16,9 +24,15 @@ type Spec struct {
 
 // Validate checks an asset ID is valid, but does not attempt to fetch
 // the asset or verify the integrity of its hash
-func (a Spec) Validate() error {
+func (a *Spec) Validate() error {
 	if a.Name == "" {
 		return errors.New("asset name cannot be empty")
 	}
+	if a.SHA512 == "" {
+		return errors.New("asset sha cannot be empty")
+	}
+	if _, err := url.Parse(a.URL); err != nil {
+		return fmt.Errorf("could not parse url %s: %s", a.URL, err)
+	}
 	return nil
 }

pkg/receiver/jobreceiver/asset/doc.go

@@ -0,0 +1,4 @@
+// Package asset facilitates the retrieval of remote runtime assets.
+//
+// Largely inspired and adapted from the source of `github.com/sensu/sensu-go`.
+package asset

pkg/receiver/jobreceiver/asset/environment.go

@@ -0,0 +1,84 @@
+package asset
+
+import (
+	"os"
+	"regexp"
+	"sort"
+	"strings"
+)
+
+var (
+	keyRegex          = regexp.MustCompile("[^a-zA-Z0-9]+")
+	pathListSeparator = string(os.PathListSeparator)
+)
+
+// key takes a string and converts it to an POSIX compliant environment key
+// variable in uppercase
+func key(s string) string {
+	return strings.ToUpper(keyRegex.ReplaceAllString(s, "_"))
+}
+
+// mergeEnvironments merges one or more sets of environment variables,
+// overwriting any existing variable in the preceding set, except for the
+// "special" variables PATH, CPATH and LD_LIBRARY_PATH.
+//
+// The "special" variables PATH, CPATH and LD_LIBRARY_PATH are merged by
+// prepending the values from right to those in left, effectively giving
+// priority to the values from right.
+//
+// The expected format for an environment variable definition is VAR=VALUE. Any
+// malformed environment variable definition will be discarded by the merge.
+func mergeEnvironments(ea []string, es ...[]string) []string {
+	envs := toMap(ea)
+
+	for i := range es {
+		env := toMap(es[i])
+		for k, v := range env {
+			switch k {
+			case "PATH", "CPATH", "LD_LIBRARY_PATH":
+				envs[k] = strings.Join([]string{v, envs[k]}, pathListSeparator)
+			default:
+				envs[k] = v
+			}
+		}
+	}
+
+	return fromMap(envs)
+}
+
+func toMap(s []string) map[string]string {
+	m := map[string]string{}
+
+	for _, v := range s {
+		// Try to split the variable definition into exactly 2 substrings:
+		// what's left of the first '=' (the variable name) and what's right
+		// of it (the variable value)
+		split := strings.SplitN(v, "=", 2)
+
+		switch len(split) {
+		case 1:
+			if split[0] != v {
+				// We came across VAR=, which is equivalent to VAR=""
+				m[split[0]] = ""
+			}
+		case 2:
+			// See _windows.go
+			key := coerceKey(split[0])
+			// A proper VAR=VALUE definition
+			m[key] = split[1]
+		}
+	}
+
+	return m
+}
+
+func fromMap(m map[string]string) []string {
+	s := []string{}
+
+	for k, v := range m {
+		s = append(s, k+"="+v)
+	}
+	sort.StringSlice(s).Sort()
+
+	return s
+}

pkg/receiver/jobreceiver/asset/environment_posix.go

@@ -0,0 +1,10 @@
+//go:build !windows
+// +build !windows
+
+package asset
+
+// POSIX compliant platforms use case-sensitive variables, no coercion
+// required.
+func coerceKey(k string) string {
+	return k
+}

pkg/receiver/jobreceiver/asset/environment_posix_test.go

@@ -0,0 +1,35 @@
+//go:build !windows
+// +build !windows
+
+package asset
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestMergeEnvironmentsPosix(t *testing.T) {
+	cases := []struct {
+		name     string
+		env1     []string
+		env2     []string
+		env3     []string
+		expected []string
+	}{
+		{
+			name:     "mixed case",
+			env1:     []string{"VAR1=VALUE1"},
+			env2:     []string{"VAR2=VALUE2"},
+			env3:     []string{"Var1=VALUE3", "Var2=VALUE4"},
+			expected: []string{"VAR1=VALUE1", "VAR2=VALUE2", "Var1=VALUE3", "Var2=VALUE4"},
+		},
+	}
+
+	for _, tt := range cases {
+		t.Run(tt.name, func(t *testing.T) {
+			result := mergeEnvironments(tt.env1, tt.env2, tt.env3)
+			assert.ElementsMatch(t, result, tt.expected)
+		})
+	}
+}

pkg/receiver/jobreceiver/asset/environment_test.go

@@ -0,0 +1,132 @@
+package asset
+
+import (
+	"fmt"
+	"os"
+	"strings"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+// Makes platform compliant list of values
+func mkList(key string, s ...string) string {
+	val := strings.Join(s, string(os.PathListSeparator))
+	return fmt.Sprintf("%s=%s", key, val)
+}
+
+func TestMergeEnvironments(t *testing.T) {
+	cases := []struct {
+		name     string
+		env1     []string
+		env2     []string
+		env3     []string
+		expected []string
+	}{
+		{
+			name:     "Empty + Empty = Empty",
+			env1:     []string{},
+			env2:     []string{},
+			expected: []string{},
+		},
+		{
+			name:     "right identity",
+			env1:     []string{"VAR1=VALUE1", "VAR2=VALUE2"},
+			env2:     []string{},
+			expected: []string{"VAR1=VALUE1", "VAR2=VALUE2"},
+		},
+		{
+			name:     "left identity",
+			env1:     []string{},
+			env2:     []string{"VAR1=VALUE1", "VAR2=VALUE2"},
+			expected: []string{"VAR1=VALUE1", "VAR2=VALUE2"},
+		},
+		{
+			name:     "no overlap",
+			env1:     []string{"VAR1=VALUE1", "VAR2=VALUE2"},
+			env2:     []string{"VAR3=VALUE3"},
+			expected: []string{"VAR1=VALUE1", "VAR2=VALUE2", "VAR3=VALUE3"},
+		},
+		{
+			name:     "overlap",
+			env1:     []string{"VAR1=VALUE1", "VAR2=VALUE2"},
+			env2:     []string{"VAR1=VALUE3", "VAR2=VALUE4"},
+			expected: []string{"VAR1=VALUE3", "VAR2=VALUE4"},
+		},
+		{
+			name:     "PATH merge",
+			env1:     []string{mkList("PATH", "c", "d")},
+			env2:     []string{mkList("PATH", "a", "b")},
+			expected: []string{mkList("PATH", "a", "b", "c", "d")},
+		},
+		{
+			name:     "CPATH merge",
+			env1:     []string{mkList("CPATH", "c", "d")},
+			env2:     []string{mkList("CPATH", "a", "b")},
+			expected: []string{mkList("CPATH", "a", "b", "c", "d")},
+		},
+		{
+			name:     "LD_LIBRARY_PATH merge",
+			env1:     []string{mkList("LD_LIBRARY_PATH", "c", "d")},
+			env2:     []string{mkList("LD_LIBRARY_PATH", "a", "b")},
+			expected: []string{mkList("LD_LIBRARY_PATH", "a", "b", "c", "d")},
+		},
+		{
+			name:     "complex example",
+			env1:     []string{"VAR1=VALUE1", mkList("PATH", "/bin", "/sbin")},
+			env2:     []string{mkList("PATH", "~/bin", "~/.local/bin"), "VAR2=VALUE2"},
+			expected: []string{"VAR1=VALUE1", "VAR2=VALUE2", mkList("PATH", "~/bin", "~/.local/bin", "/bin", "/sbin")},
+		},
+		{
+			name:     "discard invalid environment variables",
+			env1:     []string{"VAR1", "VAR2=VALUE2", "garbagelol"},
+			env2:     []string{"VAR3="},
+			expected: []string{"VAR2=VALUE2", "VAR3="},
+		},
+		{
+			name:     "more than two sets of variables",
+			env1:     []string{mkList("CPATH", "e", "f"), "VAR1=two"},
+			env2:     []string{mkList("CPATH", "c", "d"), "VAR1=one"},
+			env3:     []string{mkList("CPATH", "a", "b")},
+			expected: []string{mkList("CPATH", "a", "b", "c", "d", "e", "f"), "VAR1=one"},
+		},
+	}
+
+	for _, tt := range cases {
+		t.Run(tt.name, func(t *testing.T) {
+			result := mergeEnvironments(tt.env1, tt.env2, tt.env3)
+			assert.ElementsMatch(t, result, tt.expected)
+		})
+	}
+}
+
+func TestKey(t *testing.T) {
+	tests := []struct {
+		name string
+		s    string
+		want string
+	}{
+		{
+			name: "special characters are replaced",
+			s:    "FOO@BAR",
+			want: "FOO_BAR",
+		},
+		{
+			name: "the key is uppercase",
+			s:    "foo",
+			want: "FOO",
+		},
+		{
+			name: "underscores are preserved",
+			s:    "FOO_BAR",
+			want: "FOO_BAR",
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if got := key(tt.s); got != tt.want {
+				t.Errorf("Key() = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}

pkg/receiver/jobreceiver/asset/environment_windows.go

@@ -0,0 +1,13 @@
+//go:build windows
+// +build windows
+
+package asset
+
+import "strings"
+
+// On Windows, environment variables are case-insensitive; to avoid conflicts we
+// coerce all keys to UPPER CASE.
+// https://docs.microsoft.com/en-us/dotnet/api/system.environment.getenvironmentvariable?view=netframework-4.7.2
+func coerceKey(k string) string {
+	return strings.ToUpper(k)
+}

pkg/receiver/jobreceiver/asset/environment_windows_test.go

@@ -0,0 +1,35 @@
+//go:build windows
+// +build windows
+
+package asset
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestMergeEnvironmentsWindows(t *testing.T) {
+	cases := []struct {
+		name     string
+		env1     []string
+		env2     []string
+		env3     []string
+		expected []string
+	}{
+		{
+			name:     "mixed case",
+			env1:     []string{"VAR1=VALUE1"},
+			env2:     []string{"VAR2=VALUE2"},
+			env3:     []string{"Var1=VALUE3", "Var2=VALUE4"},
+			expected: []string{"VAR1=VALUE3", "VAR2=VALUE4"},
+		},
+	}
+
+	for _, tt := range cases {
+		t.Run(tt.name, func(t *testing.T) {
+			result := mergeEnvironments(tt.env1, tt.env2, tt.env3)
+			assert.ElementsMatch(t, result, tt.expected)
+		})
+	}
+}