feat: implement analytics recoder #4388

frrist · 2024-09-09T22:14:09Z

No description provided.

wdbaruni

Looks good overall

wdbaruni · 2024-09-12T06:38:00Z

pkg/analytics/analytics.go

+
+func New(ctx context.Context, opts ...Option) (*LogRecorder, error) {
+	config := &Config{
+		otlpEndpoint: "localhost:4317", // Default endpoint


what does it mean to have a default local endpoint?

Its the endpoint we will be sending data to, e.g. telemetry.expanso.io - this is a temporary solution for local development.

wdbaruni · 2024-09-12T06:41:13Z

pkg/analytics/analytics.go

+	if err := a.provider.ForceFlush(ctx); err != nil {
+		return fmt.Errorf("failed to flush analytics: %w", err)
+	}


will this block and delay shutdown until ctx times out?

Not sure what you mean. This function will return when ForceFlush and Shutdown return, or when the passed ctx times out.

force flush means all buffered events will be flushed to the collector before it exists, or at least thats how it sounds like. If you want to do that, then you need to define a timeout and an aggressive one as pubilshing telemetry should be best effort

wdbaruni · 2024-09-12T06:43:25Z

pkg/analytics/analytics.go

+		otellog.String(EventKey, string(event)),
+		otellog.Map(PropertiesKey, properties...),
+	)
+	a.provider.Logger("bacalhau-analytics").Emit(ctx, record)


should we create the logger in the constructor?

The logger name is used as the Scope field from an event - and an event emitted looks roughly like this:

{ "Timestamp": "2024-09-12T21:48:22.150123874Z", "ObservedTimestamp": "2024-09-12T14:48:22.150128342-07:00", "Severity": 0, "SeverityText": "", "Body": { "Type": "Empty", "Value": null }, "Attributes": [ ...<omitted>... ], "Scope": { "Name": "bacalhau-analytics", "Version": "", "SchemaURL": "" }, "DroppedAttributes": 0 }

Eventually I image we could use the logger name to differentiate between events from clients and servers, or possibly something else. For now I'll define a constant for this field (suggestions?) and create it in the constructor.

wdbaruni · 2024-09-12T06:44:51Z

pkg/analytics/analytics.go

+	record.SetTimestamp(time.Now().UTC())
+	record.AddAttributes(
+		otellog.String(EventKey, string(event)),
+		otellog.Map(PropertiesKey, properties...),


to make things flexible and support more event types with minimal effort, we can just publish properties as a json string as done in the poc https://github.com/bacalhau-project/telemetry-poc/blob/main/app/models.go#L35

though this also works and I see its value

The pattern I've implement here forces us to be specific with the fields we extract from jobs/executions/evaluations/models, removes the possibility of encountering an error while extracting fields (cuz no json marshal), and also provides a type system for each event:

{ "Key": "properties", "Value": { "Type": "Map", "Value": [ { "Key": "id", "Value": { "Type": "String", "Value": "j-67bff92a-5938-4324-9a1d-5226371e41b1" } }, { "Key": "name", "Value": { "Type": "String", "Value": "j-67bff92a-5938-4324-9a1d-5226371e41b1" } }, { "Key": "namespace", "Value": { "Type": "String", "Value": "default" } }, { "Key": "type", "Value": { "Type": "String", "Value": "batch" } }, { "Key": "count", "Value": { "Type": "Int64", "Value": 1 } }, { "Key": "state", "Value": { "Type": "String", "Value": "Completed" } },

The alternative is to implement specific types for each event as was done in the POC, e.g.

type Job struct { ID string `json:"id"` Type string `json:"type"` Image string `json:"image"` CPUCores int `json:"cpu_cores"` MemoryMB int `json:"memory_mb"` } func stubJob() Job { return Job{ ID: uuid.NewString(), Type: "docker", Image: "ubuntu:latest", CPUCores: 2, MemoryMB: 4096, } }

Which would you prefer?

Parsing and processing the first version is going to be a pain, and I don't think it will play nicely with posthog. My preference is having specific types and publish their json format. This simplies post processing as well when we just deserialize to the same type easily and process. lets just be consistent with using pascalcase everywhere

sounds good, I will create an event structure rather than use attributes.

wdbaruni · 2024-09-12T06:50:02Z

pkg/analytics/util.go

+		otellog.String("name", j.Name),
+		otellog.String("namespace", j.Namespace),


might contain PII. We are only interested in knowing if these values are defined or not. Meaning true if name doesn't equal id, and hash of namespace which we can know if they are using default or different values

Got it.
If name == id omit the name.
if name != id hash the name and include the digest.

name_set = name != id # helps understand if name is being set. don't care about its hash or how many names they have
namespace = hash(namespace) # helps with understanding how many namespaces a single network can have

wdbaruni · 2024-09-12T06:53:13Z

pkg/analytics/util.go

+		otellog.String("task_cpu", t.ResourcesConfig.CPU),
+		otellog.String("task_memory", t.ResourcesConfig.Memory),
+		otellog.String("task_disk", t.ResourcesConfig.Disk),
+		otellog.String("task_gpu", t.ResourcesConfig.GPU),


we should publish the converted resource values to have a normalized view of allocated resources. Meaning 1GB and 1024MB should be published as a single value, which should be bytes

Meaning record an integer value representing the disk and memory requirements (in bytes) of the job, yeah?

Correct. Handle failure gracefuly and set zero values instead of propagating the error

wdbaruni · 2024-09-12T06:57:19Z

pkg/jobstore/boltdb/store.go

+func WithRecorder(r analytics.Recorder) Option {
+	return func(store *BoltJobStore) {
+		store.recorder = r
+	}
+}
+


It doesn't sound right to have the recorder inside boltdb. Maybe a better place is a watcher, but the new watch library is not integrated with the job store yet. Please add a TODO to revisit this in the future

wdbaruni · 2024-09-12T06:58:41Z

pkg/jobstore/boltdb/store.go

@@ -1000,6 +1009,7 @@ func (b *BoltJobStore) updateJobState(tx *bolt.Tx, request jobstore.UpdateJobSta
 	}

 	if job.IsTerminal() {
+		b.recorder.EmitJobEvent(context.TODO(), analytics.JobComplete, job)


JobComplete is not an accurate event type as also failed and cancelled events are published in the same event, but Completed means successful compeletion in our job states. Just Job for now will do

Failed and Cancelled jobs are "complete", yea? What if we called it Job_Terminal? The nature of a jobs terminality can be gathered from inspecting its State which is included in the data we send.
By no means am I arguing against your proposal - just trying to think of a more specific name for this event which is recorded when the job reaches a terminal state.

job completed means successful completion

bacalhau/pkg/models/job.go

Lines 33 to 35 in 16538b3

// JobStateTypeCompleted is the state of a job that has successfully completed.

// Only valid for batch jobs.

JobStateTypeCompleted

I meant completed in the sense that the job is no longer running - it's stopped, terminal, etc.. I'll just call this Job

wdbaruni · 2024-09-12T09:45:33Z

cmd/cli/serve/serve.go

+		}
+	}
+
+	requesterConfig, err := GetRequesterConfig(cfg, isRequesterNode, recorder)


Unpopular opinion, it might make sense to instantiate a global analytic recorder similar to how loggers, and otel providers work. This avoids having to pass the recorder around specially that we will be emitting events in different places

I didn't use a global here since we'd like to only create a recorder if the configuration allows it.
Our logger, and other otel providers can all be configured with environment variables which makes it easier to configure them inside an init() method.
I'll leave the decision here up to you.

I know you didn't use globals in your implementation, but it is a common practice and we are already doing it today with setting otel trace and metric providers

bacalhau/pkg/telemetry/metrics.go

Line 39 in 246a2bd

otel.SetMeterProvider(meterProvider)

globals doesn't mean you set them inside an init(), but wherver you set them you then assign them to a global reference

okay I will set it as the global provider.

One issue I've encountered with this approach (using a global) is that errors are handled via https://github.com/bacalhau-project/bacalhau/blob/main/pkg/telemetry/common.go#L17 and surfaced to the user. Currently exploring options to mitigated this.

wdbaruni · 2024-09-12T10:02:36Z

pkg/analytics/analytics.go

+	a.provider.Logger("bacalhau-analytics").Emit(ctx, record)
+}
+
+func (a *LogRecorder) EmitJobEvent(ctx context.Context, event EventType, j models.Job) {


Remember we need to collect information about executions as well to understand how often failures happen before job is completed, how long jobs in the queue before execution is created, how evenly distributed node selection is, and many more

I think we can achieve this with a new event type and method, yea? Or could simply call EmitEvent which this method uses.

I am trying to simplify processing by having a single self contained event. better to have this in a single event

okay I will create an event structure rather than use attributes.

pkg/analytics/analytics.go

pkg/node/node.go

pkg/analytics/analytics.go

pkg/analytics/models.go

wdbaruni

There is a bug with logging errors. you can merge after addressing it

wdbaruni · 2024-09-17T06:55:10Z

pkg/telemetry/common.go

+			} else {
+				log.Err(err).Msg("Error occurred while handling spans")
+			}
+		default:
+			// fall through and log the error
+		}
 		log.Err(err).Msg("Error occurred while handling spans")


you are still logging the error at the end and will log the error twice for non analytics errors

feat: implement analytics recoder

60aa510

wdbaruni reviewed Sep 12, 2024

View reviewed changes

address feedback

ddaf4f7

frrist linked an issue Sep 16, 2024 that may be closed by this pull request

Develop a framework for collecting various types of telemetry #4317

Closed

frrist mentioned this pull request Sep 16, 2024

telemetry: implement tracing for job create and updates in jobstore #4232

Closed

frrist marked this pull request as ready for review September 16, 2024 17:38

Merge branch 'main' into frrist/analytics

af92d42

frrist requested a review from wdbaruni September 16, 2024 18:12

wdbaruni reviewed Sep 16, 2024

View reviewed changes

frrist added 2 commits September 16, 2024 11:52

chore: remove stdout exporter, set endpoint

b7a9d18

address feedback

0c20e08

wdbaruni approved these changes Sep 17, 2024

View reviewed changes

frrist self-assigned this Sep 17, 2024

fix logging and correct endpoint

02d8152

frrist merged commit a912d5b into main Sep 17, 2024
3 of 4 checks passed

frrist deleted the frrist/analytics branch September 17, 2024 16:15

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

feat: implement analytics recoder #4388

feat: implement analytics recoder #4388

frrist commented Sep 9, 2024

wdbaruni left a comment

wdbaruni Sep 12, 2024

frrist Sep 12, 2024

wdbaruni Sep 12, 2024

frrist Sep 12, 2024

wdbaruni Sep 12, 2024

wdbaruni Sep 12, 2024

frrist Sep 12, 2024

wdbaruni Sep 12, 2024

frrist Sep 12, 2024

wdbaruni Sep 12, 2024

frrist Sep 13, 2024

wdbaruni Sep 12, 2024

frrist Sep 12, 2024

wdbaruni Sep 12, 2024

wdbaruni Sep 12, 2024

frrist Sep 12, 2024

wdbaruni Sep 12, 2024

wdbaruni Sep 12, 2024

wdbaruni Sep 12, 2024

frrist Sep 12, 2024

wdbaruni Sep 12, 2024

frrist Sep 13, 2024

wdbaruni Sep 12, 2024

frrist Sep 12, 2024

wdbaruni Sep 12, 2024

frrist Sep 13, 2024

frrist Sep 16, 2024

wdbaruni Sep 12, 2024

frrist Sep 12, 2024 •

edited

Loading

wdbaruni Sep 12, 2024

frrist Sep 13, 2024

wdbaruni left a comment

wdbaruni Sep 17, 2024

		otellog.String("name", j.Name),
		otellog.String("namespace", j.Namespace),

	// JobStateTypeCompleted is the state of a job that has successfully completed.
	// Only valid for batch jobs.
	JobStateTypeCompleted

feat: implement analytics recoder #4388

feat: implement analytics recoder #4388

Conversation

frrist commented Sep 9, 2024

wdbaruni left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

frrist Sep 12, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

wdbaruni left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

frrist Sep 12, 2024 •

edited

Loading