Repository: osbuild/osbuild-composer
Pull request #3750 (2023-10-18 19:31:05)

Add pkg splunk_logging
This pull request includes:

- [ ] adequate testing for the new functionality or fixed issue
- [ ] adequate documentation informing people about the change such as
  - [ ] submit a PR for the [guides](https://github.com/osbuild/guides) repository if this PR changed any behavior described there: https://www.osbuild.org/guides/

<!--
Thanks for proposing a change to osbuild-composer!

Please don't remove the above check list. These are things that each pull
request must have before it is merged. It helps maintainers to not forget
anything.

If the reason for ticking any of the boxes is ambiguous, please add a short
note explaining why.

In addition, if this pull request fixes a downstream issue, please refer to
test/README.md and add these additional items:

- [ ] 1st commit of any `rhbz#` related PR contains bug reproducer; CI reports FAIL or
- [ ] PR contains automated tests for new functionality and
- [ ] QE has approved reproducer/new tests and
- [ ] Subsequent commits provide bug fixes without modifying the reproducer; CI reports PASS and
- [ ] QE approves this PR; RHBZ status is set to `MODIFIED + Verified=Tested`

Information regarding our GitLab pipeline (Schutzbot):

CI will not be ran automatically if WIP label is applied or the PR is in DRAFT state, instead only
a link will be provided to the pipeline which can then be triggered manually if desired. To run the
CI automatically either switch the PR to ready or apply WIP+test label.

Outside contributors need manual approval from one of the osbuild-composer maintainers.

Schutzbot will only be triggered if all Tests jobs in GitHub workflow succeed.
-->


croissanne: Rebased to prevent OCI private key leaking. Apologies :(

Patch:
diff --git a/pkg/splunk_logger/splunk_hook.go b/pkg/splunk_logger/splunk_hook.go
new file mode 100644
index 000000000..4c0f0702a
--- /dev/null
+++ b/pkg/splunk_logger/splunk_hook.go
@@ -0,0 +1,45 @@
+package logger
+
+import (
+	"context"
+	"fmt"
+	"os"
+
+	"github.com/sirupsen/logrus"
+)
+
+type SplunkHook struct {
+	sl *SplunkLogger
+}
+
+func NewSplunkHook(context context.Context, host, port, token, source string) (*SplunkHook, error) {
+	url := fmt.Sprintf("https://%s:%s/services/collector/event", host, port)
+	hostname, err := os.Hostname()
+	if err != nil {
+		return nil, err
+	}
+
+	return &SplunkHook{
+		sl: NewSplunkLogger(context, url, token, source, hostname),
+	}, nil
+}
+
+func (sh *SplunkHook) Fire(entry *logrus.Entry) error {
+	msg, err := entry.String()
+	if err != nil {
+		return err
+	}
+
+	return sh.sl.LogWithTime(entry.Time, msg)
+}
+
+func (sh *SplunkHook) Levels() []logrus.Level {
+	return []logrus.Level{
+		logrus.PanicLevel,
+		logrus.FatalLevel,
+		logrus.ErrorLevel,
+		logrus.WarnLevel,
+		logrus.InfoLevel,
+		logrus.DebugLevel,
+	}
+}
diff --git a/pkg/splunk_logger/splunk_logger.go b/pkg/splunk_logger/splunk_logger.go
new file mode 100644
index 000000000..1ec68a4e1
--- /dev/null
+++ b/pkg/splunk_logger/splunk_logger.go
@@ -0,0 +1,154 @@
+package logger
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"fmt"
+	"net/http"
+	"os"
+	"time"
+
+	"github.com/hashicorp/go-retryablehttp"
+)
+
+const (
+	PayloadsChannelSize = 1000
+	// in seconds, how often batched events should be sent
+	SendFrequency = 5
+)
+
+type SplunkLogger struct {
+	client   *http.Client
+	url      string
+	token    string
+	source   string
+	hostname string
+
+	payloads chan *SplunkPayload
+}
+
+type SplunkPayload struct {
+	// splunk expects unix time in seconds
+	Time  int64       `json:"time"`
+	Host  string      `json:"host"`
+	Event SplunkEvent `json:"event"`
+}
+
+type SplunkEvent struct {
+	Message string `json:"message"`
+	Ident   string `json:"ident"`
+	Host    string `json:"host"`
+}
+
+func NewSplunkLogger(context context.Context, url, token, source, hostname string) *SplunkLogger {
+	sl := &SplunkLogger{
+		client:   retryablehttp.NewClient().StandardClient(),
+		url:      url,
+		token:    token,
+		source:   source,
+		hostname: hostname,
+	}
+
+	ticker := time.NewTicker(time.Second * SendFrequency)
+	sl.payloads = make(chan *SplunkPayload, PayloadsChannelSize)
+
+	go sl.flushPayloads(context, ticker.C)
+
+	return sl
+}
+
+func (sl *SplunkLogger) flushPayloads(context context.Context, ticker <-chan time.Time) {
+	var payloads []*SplunkPayload
+	for {
+		select {
+		case <-context.Done():
+			err := sl.SendPayloads(payloads)
+			if err != nil {
+				fmt.Fprintf(os.Stderr, "Splunk logger unable to send payloads: %v", err)
+			}
+			return
+		case p := <-sl.payloads:
+			if p != nil {
+				payloads = append(payloads, p)
+			}
+			if len(payloads) == PayloadsChannelSize {
+				err := sl.SendPayloads(payloads)
+				if err != nil {
+					fmt.Fprintf(os.Stderr, "Splunk logger unable to send payloads: %v", err)
+				}
+				payloads = nil
+			}
+		case <-ticker:
+			err := sl.SendPayloads(payloads)
+			if err != nil {
+				fmt.Fprintf(os.Stderr, "Splunk logger unable to send payloads: %v", err)
+			}
+			payloads = nil
+		}
+	}
+}
+
+func (sl *SplunkLogger) SendPayloads(payloads []*SplunkPayload) error {
+	if len(payloads) == 0 {
+		return nil
+	}
+
+	buf := bytes.NewBuffer(nil)
+	for _, pl := range payloads {
+		b, err := json.Marshal(pl)
+		if err != nil {
+			return err
+		}
+
+		_, err = buf.Write(b)
+		if err != nil {
+			return err
+		}
+	}
+
+	req, err := http.NewRequest("POST", sl.url, bytes.NewReader(buf.Bytes()))
+	if err != nil {
+		return err
+	}
+	req.Header.Add("Content-Type", "application/json")
+	req.Header.Add("Authorization", fmt.Sprintf("Splunk %s", sl.token))
+
+	res, err := sl.client.Do(req)
+	if err != nil {
+		return err
+	}
+	defer func() {
+		if err := res.Body.Close(); err != nil {
+			fmt.Fprintf(os.Stderr, "Unable to close response body when sending payloads")
+		}
+	}()
+
+	if res.StatusCode != http.StatusOK {
+		buf := bytes.Buffer{}
+		_, err = buf.ReadFrom(res.Body)
+		if err != nil {
+			return fmt.Errorf("Error forwarding to splunk: parsing response failed: %v", err)
+		}
+		return fmt.Errorf("Error forwarding to splunk: %s", buf.String())
+	}
+	return nil
+}
+
+func (sl *SplunkLogger) LogWithTime(t time.Time, msg string) error {
+	sp := SplunkPayload{
+		Time: t.Unix(),
+		Host: sl.hostname,
+		Event: SplunkEvent{
+			Message: msg,
+			Ident:   sl.source,
+			Host:    sl.hostname,
+		},
+	}
+	select {
+	case sl.payloads <- &sp:
+	default:
+		return fmt.Errorf("Error queueing splunk payload, channel full")
+	}
+	return nil
+}


Test patch:
diff --git a/pkg/splunk_logger/splunk_logger_test.go b/pkg/splunk_logger/splunk_logger_test.go
new file mode 100644
index 000000000..fab4dee07
--- /dev/null
+++ b/pkg/splunk_logger/splunk_logger_test.go
@@ -0,0 +1,65 @@
+package logger
+
+import (
+	"context"
+	"encoding/json"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestSplunkLogger(t *testing.T) {
+	ch := make(chan bool)
+	time.AfterFunc(time.Second*10, func() {
+		ch <- false
+	})
+	count := 0
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		// make sure the logger retries requests
+		if count == 0 {
+			count += 1
+			w.WriteHeader(http.StatusInternalServerError)
+			return
+		}
+		require.Equal(t, "Splunk", r.Header.Get("Authorization"))
+		require.Equal(t, "application/json", r.Header.Get("Content-Type"))
+		var sp SplunkPayload
+		err := json.NewDecoder(r.Body).Decode(&sp)
+		require.NoError(t, err)
+		require.Equal(t, "test-host", sp.Host)
+		require.Equal(t, "test-host", sp.Event.Host)
+		require.Equal(t, "image-builder", sp.Event.Ident)
+		require.Equal(t, "message", sp.Event.Message)
+		ch <- true
+	}))
+	sl := NewSplunkLogger(context.Background(), srv.URL, "", "image-builder", "test-host")
+	require.NoError(t, sl.LogWithTime(time.Now(), "message"))
+	require.True(t, <-ch)
+}
+
+func TestSplunkLoggerContext(t *testing.T) {
+	ch := make(chan bool)
+	time.AfterFunc(time.Second*10, func() {
+		ch <- false
+	})
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		require.Equal(t, "Splunk", r.Header.Get("Authorization"))
+		require.Equal(t, "application/json", r.Header.Get("Content-Type"))
+		var sp SplunkPayload
+		err := json.NewDecoder(r.Body).Decode(&sp)
+		require.NoError(t, err)
+		require.Equal(t, "test-host", sp.Host)
+		require.Equal(t, "test-host", sp.Event.Host)
+		require.Equal(t, "image-builder", sp.Event.Ident)
+		require.Equal(t, "message", sp.Event.Message)
+		ch <- true
+	}))
+	ctx, cancel := context.WithTimeout(context.Background(), time.Second*1)
+	defer cancel()
+	sl := NewSplunkLogger(ctx, srv.URL, "", "image-builder", "test-host")
+	require.NoError(t, sl.LogWithTime(time.Now(), "message"))
+	require.True(t, <-ch)
+}