ship logs to http endpoint

cmd/launcher/launcher.go

@@ -39,6 +39,8 @@ import (
 	"github.com/kolide/launcher/pkg/debug"
 	"github.com/kolide/launcher/pkg/launcher"
 	"github.com/kolide/launcher/pkg/log/checkpoint"
+	"github.com/kolide/launcher/pkg/log/logshipper"
+	"github.com/kolide/launcher/pkg/log/teelogger"
 	"github.com/kolide/launcher/pkg/osquery"
 	osqueryInstanceHistory "github.com/kolide/launcher/pkg/osquery/runtime/history"
 	"github.com/kolide/launcher/pkg/service"
@@ -144,6 +146,19 @@ func runLauncher(ctx context.Context, cancel func(), opts *launcher.Options) err
 	flagController := flags.NewFlagController(logger, stores[storage.AgentFlagsStore], fcOpts...)
 	k := knapsack.New(stores, flagController, db)
 
+	// Need to set up the log shipper so that we can get the logger early
+	// and pass it to the various systems.
+	isShippingLogs := k.ObservabilityIngestServerURL() != ""
+	var logShipper *logshipper.LogShipper
+	if isShippingLogs {
+		logShipper, err = logshipper.New(k)
+		if err != nil {
+			logger.Log("msg", "failed to create log shipper", "err", err)
+		} else {
+			logger = teelogger.New(logger, logShipper)
+		}
+	}
+
 	// construct the appropriate http client based on security settings
 	httpClient := http.DefaultClient
 	if k.InsecureTLS() {
@@ -305,6 +320,13 @@ func runLauncher(ctx context.Context, cancel func(), opts *launcher.Options) err
 			runGroup.Add(exp.Execute, exp.Interrupt)
 			controlService.RegisterSubscriber(authTokensSubsystemName, exp)
 		}
+
+		// begin log shipping and subsribe to token updates
+		// nil check incase it failed to create for some reason
+		if isShippingLogs && logShipper != nil {
+			runGroup.Add(logShipper.Run, logShipper.Stop)
+			controlService.RegisterSubscriber(authTokensSubsystemName, logShipper)
+		}
 	}
 
 	runEECode := k.ControlServerURL() != "" || k.IAmBreakingEELicense()

cmd/launcher/options.go

@@ -84,7 +84,8 @@ func parseOptions(subcommandName string, args []string) (*launcher.Options, erro
 		flConfigFilePath         = flagset.String("config", defaultConfigFilePath, "config file to parse options from (optional)")
 		flExportTraces           = flagset.Bool("export_traces", false, "Whether to export traces")
 		flTraceSamplingRate      = flagset.Float64("trace_sampling_rate", 0.0, "What fraction of traces should be sampled")
-		flIngestServerURL        = flagset.String("ingest_url", "", "Where to export traces and logs")
+		flIngestServerURL        = flagset.String("ingest_url", "", "Where to export logs and other observability data")
+		flTraceIngestServerURL   = flagset.String("trace_ingest_url", "", "Where to export traces")
 		flDisableIngestTLS       = flagset.Bool("disable_ingest_tls", false, "Disable TLS for observability ingest server communication")
 
 		// osquery TLS endpoints
@@ -249,6 +250,7 @@ func parseOptions(subcommandName string, args []string) (*launcher.Options, erro
 		EnrollSecretPath:                   *flEnrollSecretPath,
 		ExportTraces:                       *flExportTraces,
 		ObservabilityIngestServerURL:       *flIngestServerURL,
+		TraceIngestServerURL:               *flTraceIngestServerURL,
 		DisableObservabilityIngestTLS:      *flDisableIngestTLS,
 		AutoloadedExtensions:               flAutoloadedExtensions,
 		IAmBreakingEELicense:               *flIAmBreakingEELicense,

pkg/agent/flags/flag_controller.go

@@ -474,6 +474,15 @@ func (fc *FlagController) ObservabilityIngestServerURL() string {
 	).get(fc.getControlServerValue(keys.ObservabilityIngestServerURL))
 }
 
+func (fc *FlagController) SetTraceIngestServerURL(url string) error {
+	return fc.setControlServerValue(keys.TraceIngestServerURL, []byte(url))
+}
+func (fc *FlagController) TraceIngestServerURL() string {
+	return NewStringFlagValue(
+		WithDefaultString(fc.cmdLineOpts.TraceIngestServerURL),
+	).get(fc.getControlServerValue(keys.TraceIngestServerURL))
+}
+
 func (fc *FlagController) SetDisableObservabilityIngestTLS(enabled bool) error {
 	return fc.setControlServerValue(keys.DisableObservabilityIngestTLS, boolToBytes(enabled))
 }

pkg/agent/flags/keys/keys.go

@@ -45,6 +45,7 @@ const (
 	ExportTraces                  FlagKey = "export_traces"
 	TraceSamplingRate             FlagKey = "trace_sampling_rate"
 	ObservabilityIngestServerURL  FlagKey = "ingest_url"
+	TraceIngestServerURL          FlagKey = "trace_ingest_url"
 	DisableObservabilityIngestTLS FlagKey = "disable_ingest_tls"
 )
 

pkg/agent/knapsack/knapsack.go

@@ -369,6 +369,13 @@ func (k *knapsack) ObservabilityIngestServerURL() string {
 	return k.flags.ObservabilityIngestServerURL()
 }
 
+func (k *knapsack) SetTraceIngestServerURL(url string) error {
+	return k.flags.SetTraceIngestServerURL(url)
+}
+func (k *knapsack) TraceIngestServerURL() string {
+	return k.flags.ObservabilityIngestServerURL()
+}
+
 func (k *knapsack) SetDisableObservabilityIngestTLS(enabled bool) error {
 	return k.flags.SetDisableObservabilityIngestTLS(enabled)
 }

pkg/agent/types/flags.go

@@ -177,10 +177,14 @@ type Flags interface {
 	SetTraceSamplingRate(rate float64) error
 	TraceSamplingRate() float64
 
-	// ObservabilityIngestServerURL is the URL of the ingest server for logs and traces
+	// ObservabilityIngestServerURL is the URL of the ingest server for logs and other observability data
 	SetObservabilityIngestServerURL(url string) error
 	ObservabilityIngestServerURL() string
 
+	// TraceIngestServerURL is the URL of the ingest server for traces
+	SetTraceIngestServerURL(url string) error
+	TraceIngestServerURL() string
+
 	// DisableObservabilityIngestTLS disables TLS for observability ingest server communication
 	SetDisableObservabilityIngestTLS(enabled bool) error
 	DisableObservabilityIngestTLS() bool

pkg/launcher/options.go

@@ -107,8 +107,10 @@ type Options struct {
 	ExportTraces bool
 	// TraceSamplingRate is a number between 0.0 and 1.0 that indicates what fraction of traces should be sampled.
 	TraceSamplingRate float64
-	// ObservabilityIngestServerURL is the URL that traces and logs will be exported to
+	// ObservabilityIngestServerURL is the URL that logs and other observability data will be exported to
 	ObservabilityIngestServerURL string
+	// TraceIngestServerURL is the URL that traces will be exported to
+	TraceIngestServerURL string
 	// DisableObservabilityIngestTLS allows for disabling TLS when connecting to the observability ingest server
 	DisableObservabilityIngestTLS bool
 

pkg/log/logshipper/authedhttpsender.go

@@ -0,0 +1,49 @@
+package logshipper
+
+import (
+	"fmt"
+	"io"
+	"net/http"
+	"time"
+)
+
+type authedHttpSender struct {
+	endpoint  string
+	authtoken string
+	client    *http.Client
+}
+
+func newAuthHttpSender(endpoint, authtoken string) *authedHttpSender {
+	return &authedHttpSender{
+		endpoint:  endpoint,
+		authtoken: authtoken,
+		client: &http.Client{
+			Timeout: 30 * time.Second,
+		},
+	}
+}
+
+func (a *authedHttpSender) Send(r io.Reader) error {
+	req, err := http.NewRequest("POST", a.endpoint, r)
+	if err != nil {
+		return err
+	}
+	req.Header.Set("Content-Type", "application/octet-stream")
+	req.Header.Set("authorization", a.authtoken)
+
+	resp, err := a.client.Do(req)
+	if err != nil {
+		return err
+	}
+	defer resp.Body.Close()
+
+	if resp.StatusCode >= 300 {
+		bodyData, err := io.ReadAll(resp.Request.Body)
+		if err != nil {
+			return fmt.Errorf("received non 200 http status code: %d, error reading body response body %w", resp.StatusCode, err)
+		}
+
+		return fmt.Errorf("received non 200 http status code: %d, response body: %s", resp.StatusCode, bodyData)
+	}
+	return nil
+}

pkg/log/logshipper/authedhttpsender_test.go

@@ -0,0 +1,47 @@
+package logshipper
+
+import (
+	"bytes"
+	"net/http"
+	"net/http/httptest"
+	"sync"
+	"testing"
+
+	"github.com/kolide/kit/ulid"
+	"github.com/stretchr/testify/require"
+)
+
+func Test_authedHttpSender_Send(t *testing.T) {
+	t.Parallel()
+	tests := []struct {
+		name string
+	}{
+		{
+			name: "happy path",
+		},
+	}
+	for _, tt := range tests {
+		tt := tt
+		t.Run(tt.name, func(t *testing.T) {
+			t.Parallel()
+
+			dataToSend := []byte(ulid.New())
+			token := ulid.New()
+
+			wg := sync.WaitGroup{}
+			wg.Add(1)
+			// create http test server with handle func that returns 200 OK
+			ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+				require.Equal(t, token, r.Header.Get("authorization"))
+				w.WriteHeader(http.StatusOK)
+				wg.Done()
+			}))
+			defer ts.Close()
+
+			authedSender := newAuthHttpSender(ts.URL, token)
+			authedSender.Send(bytes.NewBuffer(dataToSend))
+
+			wg.Wait()
+		})
+	}
+}