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,14 @@ 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.
+	var logShipper *logshipper.LogShipper
+	if k.ControlServerURL() != "" {
+		logShipper = logshipper.New(k)
+		logger = teelogger.New(logger, logShipper.Logger())
+	}
+
 	// construct the appropriate http client based on security settings
 	httpClient := http.DefaultClient
 	if k.InsecureTLS() {
@@ -305,6 +315,10 @@ 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
+		runGroup.Add(logShipper.StartShipping, logShipper.StopShipping)
+		controlService.RegisterSubscriber(authTokensSubsystemName, logShipper)
 	}
 
 	runEECode := k.ControlServerURL() != "" || k.IAmBreakingEELicense()

pkg/log/logshipper/authedhttpsender.go

@@ -0,0 +1,44 @@
+package logshipper
+
+import (
+	"fmt"
+	"io"
+	"net/http"
+)
+
+type authedHttpSender struct {
+	endpoint  string
+	authtoken string
+}
+
+func newAuthHttpSender(endpoint, authtoken string) *authedHttpSender {
+	return &authedHttpSender{
+		endpoint:  endpoint,
+		authtoken: authtoken,
+	}
+}
+
+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", fmt.Sprintf("Bearer %s", a.authtoken))
+
+	resp, err := http.DefaultClient.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/logger.go

@@ -0,0 +1,43 @@
+package logshipper
+
+import (
+	"fmt"
+	"io"
+
+	"github.com/go-kit/kit/log"
+)
+
+const (
+	truncatedFormatString = "%s[TRUNCATED]"
+)
+
+type logger struct {
+	logger log.Logger
+}
+
+func newLogger(w io.Writer) *logger {
+	return &logger{
+		logger: log.NewJSONLogger(w),
+	}
+}
+
+func (h *logger) Log(keyvals ...interface{}) error {
+	filterResults(keyvals...)
+	return h.logger.Log(keyvals...)
+}
+
+// filterResults filteres out the osquery results,
+// which just make a lot of noise in our debug logs.
+// It's a bit fragile, since it parses keyvals, but
+// hopefully that's good enough
+func filterResults(keyvals ...interface{}) {
+	// Consider switching on `method` as well?
+	for i := 0; i < len(keyvals); i += 2 {
+		if keyvals[i] == "results" && len(keyvals) > i+1 {
+			str, ok := keyvals[i+1].(string)
+			if ok && len(str) > 100 {
+				keyvals[i+1] = fmt.Sprintf(truncatedFormatString, str[0:99])
+			}
+		}
+	}
+}

pkg/log/logshipper/logshipper.go

@@ -0,0 +1,70 @@
+package logshipper
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	"github.com/go-kit/kit/log"
+	"github.com/kolide/launcher/pkg/agent/types"
+	"github.com/kolide/launcher/pkg/sendbuffer"
+)
+
+// TODO: consolidate this var, but where?
+var observabilityIngestTokenKey = []byte("observability_ingest_auth_token")
+
+type LogShipper struct {
+	sender     *authedHttpSender
+	sendBuffer *sendbuffer.SendBuffer
+	logger     *logger
+	knapsack   types.Knapsack
+}
+
+func New(k types.Knapsack) *LogShipper {
+	token, _ := k.TokenStore().Get(observabilityIngestTokenKey)
+	sender := newAuthHttpSender(logEndpoint(k.ObservabilityIngestServerURL()), string(token))
+
+	sendInterval := time.Minute * 5
+	if k.Debug() {
+		sendInterval = time.Second * 1
+	}
+
+	sendBuffer := sendbuffer.New(sender, sendbuffer.WithSendInterval(sendInterval))
+	logger := newLogger(sendBuffer)
+
+	return &LogShipper{
+		sender:     sender,
+		sendBuffer: sendBuffer,
+		logger:     logger,
+		knapsack:   k,
+	}
+}
+
+func (ls *LogShipper) Logger() log.Logger {
+	return ls.logger
+}
+
+func (ls *LogShipper) Ping() {
+	token, _ := ls.knapsack.TokenStore().Get(observabilityIngestTokenKey)
+	ls.sender.authtoken = string(token)
+	ls.sender.endpoint = logEndpoint(ls.knapsack.ObservabilityIngestServerURL())
+}
+
+// StartShipping is a no-op -- the exporter is already running in the background. The TraceExporter
+// otherwise only responds to control server events.
+func (ls *LogShipper) StartShipping() error {
+	ls.sendBuffer.StartSending()
+
+	// nothing else to do, wait for launcher to exit
+	<-context.Background().Done()
+	return nil
+}
+
+func (t *LogShipper) StopShipping(_ error) {
+	t.sendBuffer.StopSending()
+}
+
+func logEndpoint(url string) string {
+	// TODO: add new value for logging endpoint
+	return fmt.Sprintf("http://localhost:8080/log")
+}