Stream agent logs remotely

go.mod

@@ -592,7 +592,7 @@ replace (
 	// TODO: Update when https://github.com/grafana-operator/grafana-operator/pull/1301/files is merged
 	github.com/grafana-operator/grafana-operator/v5 => github.com/jaehnri/grafana-operator/v5 v5.0.0-20231107224428-8e66ce068452
 	// https://github.com/hashicorp/go-plugin/pull/251
-	github.com/hashicorp/go-plugin => github.com/alexandreLamarre/go-plugin v0.1.1-0.20230417174342-eab684801be5
+	github.com/hashicorp/go-plugin => github.com/alexandreLamarre/go-plugin v0.0.0-20231106165427-ede0fcce2acc
 	github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza => github.com/rancher-sandbox/otel-collector-contrib/pkg/stanza v0.0.0-20231108205431-9f8e6699a9d4
 	github.com/opensearch-project/opensearch-go/v2 => github.com/dbason/opensearch-go/v2 v2.0.0-20221202021211-6aec8f80bc41
 	github.com/rancher/charts-build-scripts => github.com/dbason/charts-build-scripts v0.3.4-0.20220429024555-807c076e8116

go.sum

@@ -545,8 +545,8 @@ github.com/alecthomas/units v0.0.0-20190924025748-f65c72e2690d/go.mod h1:rBZYJk5
 github.com/alecthomas/units v0.0.0-20210208195552-ff826a37aa15/go.mod h1:OMCwj8VM1Kc9e19TLln2VL61YJF0x1XFtfdL4JdbSyE=
 github.com/alecthomas/units v0.0.0-20211218093645-b94a6e3cc137 h1:s6gZFSlWYmbqAuRjVTiNNhvNRfY2Wxp9nhfyel4rklc=
 github.com/alecthomas/units v0.0.0-20211218093645-b94a6e3cc137/go.mod h1:OMCwj8VM1Kc9e19TLln2VL61YJF0x1XFtfdL4JdbSyE=
-github.com/alexandreLamarre/go-plugin v0.1.1-0.20230417174342-eab684801be5 h1:VByWr0MtaW/iS+MncS+5E8lCi2Hgc3JHVnMD6bCCwEk=
-github.com/alexandreLamarre/go-plugin v0.1.1-0.20230417174342-eab684801be5/go.mod h1:6/1TEzT0eQznvI/gV2CM29DLSkAK/e58mUWKVsPaph0=
+github.com/alexandreLamarre/go-plugin v0.0.0-20231106165427-ede0fcce2acc h1:b7BMoL6UF6kRTW3apg0XJmIG039TExK9arOKSOIY7hA=
+github.com/alexandreLamarre/go-plugin v0.0.0-20231106165427-ede0fcce2acc/go.mod h1:1BphuLk8r4UASxJBCggGswD3Wn84QLOBANVCAdNoeoY=
 github.com/alicebob/gopher-json v0.0.0-20230218143504-906a9b012302 h1:uvdUDbHQHO85qeSydJtItA4T55Pw6BtAejd0APRJOCE=
 github.com/alicebob/gopher-json v0.0.0-20230218143504-906a9b012302/go.mod h1:SGnFV6hVsYE877CKEZ6tDNTjaSXYUk6QqoIK6PrAtcc=
 github.com/alicebob/miniredis/v2 v2.30.4 h1:8S4/o1/KoUArAGbGwPxcwf0krlzceva2XVOSchFS7Eo=

internal/alerting/syncer/syncer_server.go

@@ -64,7 +64,7 @@ func NewAlertingSyncerV1(
 	init.Store(false)
 	server := &AlertManagerSyncerV1{
 		serverConfig: serverConfig,
-		lg:           logger.NewPluginLogger().WithGroup("alerting-syncer"),
+		lg:           logger.New().WithGroup("alerting-syncer"),
 		tlsConfig:    tlsConfig,
 	}
 	go func() {

pkg/agent/v1/agent.go

@@ -60,6 +60,7 @@ const (
 
 type Agent struct {
 	controlv1.UnimplementedHealthServer
+	controlv1.UnimplementedLogServer
 	AgentOptions
 	config v1beta1.AgentConfigSpec
 	router *gin.Engine

pkg/agent/v2/agent.go

@@ -24,6 +24,7 @@ import (
 	"github.com/rancher/opni/pkg/ident/identserver"
 	"github.com/rancher/opni/pkg/keyring"
 	"github.com/rancher/opni/pkg/logger"
+	"github.com/rancher/opni/pkg/logger/remotelogs"
 	"github.com/rancher/opni/pkg/machinery"
 	"github.com/rancher/opni/pkg/plugins"
 	"github.com/rancher/opni/pkg/plugins/apis/apiextensions"
@@ -118,11 +119,28 @@ func WithRebootstrap(rebootstrap bool) AgentOption {
 func New(ctx context.Context, conf *v1beta1.AgentConfig, opts ...AgentOption) (*Agent, error) {
 	options := AgentOptions{}
 	options.apply(opts...)
+
+	initCtx, initCancel := context.WithTimeout(ctx, 10*time.Second)
+	defer initCancel()
+
+	ipBuilder, err := ident.GetProviderBuilder(conf.Spec.IdentityProvider)
+	if err != nil {
+		return nil, fmt.Errorf("configuration error: %w", err)
+	}
+	ip := ipBuilder()
+	id, err := ip.UniqueIdentifier(initCtx)
+	if err != nil {
+		return nil, fmt.Errorf("error getting unique identifier: %w", err)
+	}
+
 	level := logger.DefaultLogLevel.Level()
 	if conf.Spec.LogLevel != "" {
 		level = logger.ParseLevel(conf.Spec.LogLevel)
 	}
-	lg := logger.New(logger.WithLogLevel(level)).WithGroup("agent")
+
+	ctx = logger.WithAgentId(ctx, id)
+	ctx = logger.WithMode(ctx, meta.ModeAgent)
+	lg := logger.New(logger.WithLogLevel(level), logger.WithFileWriter(logger.WriteOnlyFile(logger.GetLogFileName(id)))).WithGroup("agent")
 	lg.Debug(fmt.Sprintf("using log level: %s", level.String()))
 
 	var pl *plugins.PluginLoader
@@ -198,19 +216,6 @@ func New(ctx context.Context, conf *v1beta1.AgentConfig, opts ...AgentOption) (*
 		return nil, fmt.Errorf("agent upgrade configuration error: %w", err)
 	}
 
-	initCtx, initCancel := context.WithTimeout(ctx, 10*time.Second)
-	defer initCancel()
-
-	ipBuilder, err := ident.GetProviderBuilder(conf.Spec.IdentityProvider)
-	if err != nil {
-		return nil, fmt.Errorf("configuration error: %w", err)
-	}
-	ip := ipBuilder()
-	id, err := ip.UniqueIdentifier(initCtx)
-	if err != nil {
-		return nil, fmt.Errorf("error getting unique identifier: %w", err)
-	}
-
 	sb, err := machinery.ConfigureStorageBackend(initCtx, &conf.Spec.Storage)
 	if err != nil {
 		return nil, fmt.Errorf("error configuring keyring store broker: %w", err)
@@ -312,6 +317,14 @@ func New(ctx context.Context, conf *v1beta1.AgentConfig, opts ...AgentOption) (*
 		gatewayClient.RegisterSplicedStream(cc, md.Filename())
 	}))
 
+	ls := remotelogs.NewLogServer()
+	controlv1.RegisterLogServer(gatewayClient, ls)
+
+	pl.Hook(hooks.OnLoadMC(func(lc controlv1.LogClient, m meta.PluginMeta, cc *grpc.ClientConn) {
+		client := controlv1.NewLogClient(cc)
+		ls.AddClient(m.Filename(), client)
+	}))
+
 	return &Agent{
 		AgentOptions: options,
 		config:       conf.Spec,
@@ -395,6 +408,7 @@ func (a *Agent) ListenAndServe(ctx context.Context) error {
 			close(done)
 		}))
 
+		ctx = logger.WithAgentId(ctx, a.tenantID)
 		a.pluginLoader.LoadPlugins(ctx, a.config.PluginDir, plugins.AgentScheme,
 			plugins.WithManifest(pluginManifest),
 		)
@@ -493,6 +507,7 @@ ROUTES:
 
 func (a *Agent) runGatewayClient(ctx context.Context) error {
 	lg := a.logger
+	defer logger.CloseLogStreaming(a.tenantID)
 	isRetry := false
 	for ctx.Err() == nil {
 		if isRetry {

pkg/alerting/drivers/backend/api.go

@@ -58,7 +58,7 @@ import (
 // 		},
 // 		body:   nil,
 // 		values: nil,
-// 		logger: logger.NewPluginLogger().Named("alerting"),
+// 		logger: logger.New().Named("alerting"),
 // 	}
 // }
 

pkg/alerting/extensions/embedded_server.go

@@ -11,12 +11,13 @@ import (
 	"errors"
 	"net/http"
 
+	"log/slog"
+
 	"github.com/rancher/opni/pkg/alerting/cache"
 	"github.com/rancher/opni/pkg/alerting/extensions/destination"
 	"github.com/rancher/opni/pkg/alerting/shared"
 	alertingv1 "github.com/rancher/opni/pkg/apis/alerting/v1"
 	"github.com/rancher/opni/pkg/logger"
-	"log/slog"
 
 	// add profiles
 	_ "net/http/pprof"
@@ -64,7 +65,7 @@ func StartOpniEmbeddedServer(
 	opniAddr string,
 	sendK8s bool,
 ) *http.Server {
-	lg := logger.NewPluginLogger().WithGroup("opni.alerting")
+	lg := logger.New().WithGroup("opni.alerting")
 	es := NewEmbeddedServer(lg, 125, sendK8s)
 	mux := http.NewServeMux()
 

pkg/alerting/storage/stores.go

@@ -34,7 +34,7 @@ func NewDefaultAlertingBroker(js nats.JetStreamContext, opts ...storage_opts.Cli
 	options := &storage_opts.ClientSetOptions{}
 	options.Apply(opts...)
 	if options.Logger == nil {
-		options.Logger = logger.NewPluginLogger().WithGroup("alerting-storage-client-set")
+		options.Logger = logger.New().WithGroup("alerting-storage-client-set")
 	}
 	if options.TrackerTtl == 0 {
 		options.TrackerTtl = defaultTrackerTTLV1