offline mode

flags/flags.go

@@ -22,8 +22,8 @@ import (
 	"time"
 
 	"github.com/alecthomas/kong"
-	"go.opentelemetry.io/ebpf-profiler/tracer"
 	log "github.com/sirupsen/logrus"
+	"go.opentelemetry.io/ebpf-profiler/tracer"
 	_ "google.golang.org/grpc/encoding/proto"
 )
 
@@ -121,6 +121,8 @@ type Flags struct {
 	Hidden    FlagsHidden    `embed:"" hidden:""           prefix:""`
 
 	BPF FlagsBPF `embed:"" prefix:"bpf-"`
+
+	OfflineMode FlagsOfflineMode `embed:"" prefix:"offline-mode-"`
 }
 
 type ExitCode int
@@ -192,6 +194,14 @@ func (f Flags) Validate() ExitCode {
 		}
 	}
 
+	if len(f.OfflineMode.StoragePath) > 0 && !f.OfflineMode.Upload && (len(f.RemoteStore.Address) > 0 || len(f.OTLP.Address) > 0) {
+		return ParseError("Specified both offline mode and a remote store; this configuration is invalid.")
+	}
+
+	if f.OfflineMode.Upload && len(f.OfflineMode.StoragePath) == 0 {
+		return ParseError("Specified --offline-mode-upload without --offline-mode-storage-path.")
+	}
+
 	return ExitSuccess
 }
 
@@ -343,3 +353,9 @@ type FlagsBPF struct {
 	VerifierLogLevel uint32 `default:"0" help:"Log level of the eBPF verifier output (0,1,2). Default is 0."`
 	VerifierLogSize  int    `default:"0" help:"[deprecated] Unused."`
 }
+
+type FlagsOfflineMode struct {
+	StoragePath       string        `help:"Enables offline mode, with the data stored at the given path."`
+	RotationInterval  time.Duration `default:"10m" help:"How often to rotate and compress the offline mode log."`
+	Upload            bool          `help:"Run the uploader for data written in offline mode."`
+}

flags/grpc.go

@@ -12,10 +12,10 @@ import (
 	"github.com/grpc-ecosystem/go-grpc-middleware/v2/interceptors/logging"
 	"github.com/grpc-ecosystem/go-grpc-middleware/v2/interceptors/retry"
 	"github.com/grpc-ecosystem/go-grpc-middleware/v2/interceptors/timeout"
-	"go.opentelemetry.io/ebpf-profiler/libpf"
 	"github.com/prometheus/client_golang/prometheus"
 	log "github.com/sirupsen/logrus"
 	tracing "go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc"
+	"go.opentelemetry.io/ebpf-profiler/libpf"
 	"go.opentelemetry.io/otel/propagation"
 	"go.opentelemetry.io/otel/trace"
 	"google.golang.org/grpc"
@@ -79,7 +79,6 @@ func (f FlagsRemoteStore) setupGrpcConnection(parent context.Context, metrics *g
 		grpc.WithReturnConnectionError(),
 	}
 
-	// TLS
 	if f.Insecure {
 		opts = append(opts, grpc.WithTransportCredentials(insecure.NewCredentials()))
 	} else {

go.mod

@@ -15,12 +15,13 @@ require (
 	github.com/common-nighthawk/go-figure v0.0.0-20210622060536-734e95fb86be
 	github.com/containerd/containerd v1.7.20
 	github.com/docker/docker v27.1.1+incompatible
+	github.com/dustin/go-humanize v1.0.1
 	github.com/elastic/go-freelru v0.15.0
 	github.com/gogo/protobuf v1.3.2
 	github.com/golang/snappy v0.0.4
 	github.com/grpc-ecosystem/go-grpc-middleware/providers/prometheus v1.0.1
 	github.com/grpc-ecosystem/go-grpc-middleware/v2 v2.1.0
-	github.com/klauspost/compress v1.17.9
+	github.com/klauspost/compress v1.17.11
 	github.com/prometheus/client_golang v1.19.1
 	github.com/prometheus/common v0.54.0
 	github.com/prometheus/prometheus v0.53.1

go.sum

@@ -91,6 +91,8 @@ github.com/docker/go-events v0.0.0-20190806004212-e31b211e4f1c h1:+pKlWGMw7gf6bQ
 github.com/docker/go-events v0.0.0-20190806004212-e31b211e4f1c/go.mod h1:Uw6UezgYA44ePAFQYUehOuCzmy5zmg/+nl2ZfMWGkpA=
 github.com/docker/go-units v0.5.0 h1:69rxXcBk27SvSaaxTtLh/8llcHD8vYHT7WSdRZ/jvr4=
 github.com/docker/go-units v0.5.0/go.mod h1:fgPhTUdO+D/Jk86RDLlptpiXQzgHJF7gydDDbaIK4Dk=
+github.com/dustin/go-humanize v1.0.1 h1:GzkhY7T5VNhEkwH0PVJgjz+fX1rhBrR7pRT3mDkpeCY=
+github.com/dustin/go-humanize v1.0.1/go.mod h1:Mu1zIs6XwVuF/gI1OepvI0qD18qycQx+mFykh5fBlto=
 github.com/elastic/go-freelru v0.15.0 h1:Jo1aY8JAvpyxbTDJEudrsBfjFDaALpfVv8mxuh9sfvI=
 github.com/elastic/go-freelru v0.15.0/go.mod h1:bSdWT4M0lW79K8QbX6XY2heQYSCqD7THoYf82pT/H3I=
 github.com/elastic/go-perf v0.0.0-20241016160959-1342461adb4a h1:ymmtaN4bVCmKKeu4XEf6JEWNZKRXPMng1zjpKd+8rCU=
@@ -197,8 +199,8 @@ github.com/json-iterator/go v1.1.12 h1:PV8peI4a0ysnczrg+LtxykD8LfKY9ML6u2jnxaEnr
 github.com/json-iterator/go v1.1.12/go.mod h1:e30LSqwooZae/UwlEbR2852Gd8hjQvJoHmT4TnhNGBo=
 github.com/kisielk/errcheck v1.5.0/go.mod h1:pFxgyoBC7bSaBwPgfKdkLd5X25qrDl4LWUI2bnpBCr8=
 github.com/kisielk/gotool v1.0.0/go.mod h1:XhKaO+MFFWcvkIS/tQcRk01m1F5IRFswLeQ+oQHNcck=
-github.com/klauspost/compress v1.17.9 h1:6KIumPrER1LHsvBVuDa0r5xaG0Es51mhhB9BQB2qeMA=
-github.com/klauspost/compress v1.17.9/go.mod h1:Di0epgTjJY877eYKx5yC51cX2A2Vl2ibi7bDH9ttBbw=
+github.com/klauspost/compress v1.17.11 h1:In6xLpyWOi1+C7tXUUWv2ot1QvBjxevKAaI6IXrJmUc=
+github.com/klauspost/compress v1.17.11/go.mod h1:pMDklpSncoRMuLFrf1W9Ss9KT+0rH90U12bZKk7uwG0=
 github.com/klauspost/cpuid/v2 v2.2.8 h1:+StwCXwm9PdpiEkPyzBXIy+M9KUb4ODm0Zarf1kS5BM=
 github.com/klauspost/cpuid/v2 v2.2.8/go.mod h1:Lcz8mBdAVJIBVzewtcLocK12l3Y+JytZYpaMropDUws=
 github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE=

main.go

@@ -46,11 +46,13 @@ import (
 	"go.opentelemetry.io/otel/trace"
 	"go.opentelemetry.io/otel/trace/noop"
 	"golang.org/x/sys/unix"
+	"google.golang.org/grpc"
 
 	"github.com/parca-dev/parca-agent/analytics"
 	"github.com/parca-dev/parca-agent/config"
 	"github.com/parca-dev/parca-agent/flags"
 	"github.com/parca-dev/parca-agent/reporter"
+	"github.com/parca-dev/parca-agent/uploader"
 )
 
 var (
@@ -136,6 +138,14 @@ func mainWithExitCode() flags.ExitCode {
 		return code
 	}
 
+	if f.OfflineMode.Upload {
+		code, err := uploader.OfflineModeDoUpload(f)
+		if err != nil {
+			log.Errorf("failed to upload offline mode logs: %v", err)
+		}
+		return code
+	}
+
 	reg := prometheus.NewRegistry()
 	reg.MustRegister(
 		collectors.NewBuildInfoCollector(),
@@ -164,12 +174,17 @@ func mainWithExitCode() flags.ExitCode {
 		}
 	}
 
-	grpcConn, err := f.RemoteStore.WaitGrpcEndpoint(ctx, reg, tp)
-	if err != nil {
-		log.Errorf("failed to connect to server: %v", err)
-		return flags.ExitFailure
+	isOfflineMode := len(f.OfflineMode.StoragePath) > 0
+
+	var grpcConn *grpc.ClientConn
+	if !isOfflineMode {
+		grpcConn, err = f.RemoteStore.WaitGrpcEndpoint(ctx, reg, tp)
+		if err != nil {
+			log.Errorf("failed to connect to server: %v", err)
+			return flags.ExitFailure
+		}
+		defer grpcConn.Close()
 	}
-	defer grpcConn.Close()
 
 	presentCores, err := numcpus.GetPresent()
 	if err != nil {
@@ -291,17 +306,32 @@ func mainWithExitCode() flags.ExitCode {
 	intervals := times.New(5*time.Second, f.Profiling.Duration, f.Profiling.ProbabilisticInterval)
 	times.StartRealtimeSync(mainCtx, f.ClockSyncInterval)
 
+	var client profilestoregrpc.ProfileStoreServiceClient
+	var debuginfoClient debuginfogrpc.DebuginfoServiceClient
+	if grpcConn != nil {
+		client = profilestoregrpc.NewProfileStoreServiceClient(grpcConn)
+		debuginfoClient = debuginfogrpc.NewDebuginfoServiceClient(grpcConn)
+	}
+
+	var offlineModeConfig *reporter.OfflineModeConfig
+	if isOfflineMode {
+		offlineModeConfig = &reporter.OfflineModeConfig{
+			StoragePath:      f.OfflineMode.StoragePath,
+			RotationInterval: f.OfflineMode.RotationInterval,
+		}
+	}
+
 	// Network operations to CA start here
 	// Connect to the collection agent
 	parcaReporter, err := reporter.New(
 		memory.DefaultAllocator,
-		profilestoregrpc.NewProfileStoreServiceClient(grpcConn),
-		debuginfogrpc.NewDebuginfoServiceClient(grpcConn),
+		client,
+		debuginfoClient,
 		externalLabels,
 		f.Profiling.Duration,
 		f.Debuginfo.Strip,
 		f.Debuginfo.UploadMaxParallel,
-		f.Debuginfo.UploadDisable,
+		f.Debuginfo.UploadDisable || isOfflineMode,
 		int64(f.Profiling.CPUSamplingFrequency),
 		traceHandlerCacheSize,
 		f.Debuginfo.UploadQueueSize,
@@ -310,6 +340,7 @@ func mainWithExitCode() flags.ExitCode {
 		relabelConfigs,
 		buildInfo.VcsRevision,
 		reg,
+		offlineModeConfig,
 	)
 	if err != nil {
 		return flags.Failure("Failed to start reporting: %v", err)
@@ -413,8 +444,10 @@ func mainWithExitCode() flags.ExitCode {
 
 	log.Info("Stop processing ...")
 	rep.Stop()
-	if err := grpcConn.Close(); err != nil {
-		log.Fatalf("Stopping connection of OTLP client client failed: %v", err)
+	if grpcConn != nil {
+		if err := grpcConn.Close(); err != nil {
+			log.Fatalf("Stopping connection of OTLP client client failed: %v", err)
+		}
 	}
 
 	log.Info("Exiting ...")