gcm/slurmprocessor/factory.go at main · stephenacn/gcm · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
// Copyright (c) Meta Platforms, Inc. and affiliates.
// All rights reserved.
package slurmprocessor

import (
	"context"
	"log"

	"go.opentelemetry.io/collector/component"
	"go.opentelemetry.io/collector/consumer"
	"go.opentelemetry.io/collector/processor"
	"go.opentelemetry.io/collector/processor/processorhelper"

	shelper "github.com/facebookresearch/gcm/shelper"
)

const (
	// The value of "type" key in configuration.
	typeStr = "slurm"
)

// NewFactory creates a factory for the routing processor.
func NewFactory() processor.Factory {
	return processor.NewFactory(
		component.MustNewType(typeStr),
		createDefaultConfig,
		processor.WithTraces(createTracesProcessor, component.StabilityLevelAlpha),
		processor.WithMetrics(createMetricsProcessor, component.StabilityLevelAlpha),
		processor.WithLogs(createLogsProcessor, component.StabilityLevelAlpha),
	)
}

func createDefaultConfig() component.Config {
	return &shelper.Config{
		Source:         shelper.SlurmCtld,
		CacheDuration:  60,
		CacheFilepath:  "/tmp/slurmprocessor_cache.json",
		QuerySlurmCtld: false,
	}
}

func createTracesProcessor(
	ctx context.Context,
	set processor.Settings,
	cfg component.Config,
	nextTracesConsumer consumer.Traces,
) (processor.Traces, error) {
	log.Println("Creating Trace Processor")
	st := newSlurmInfoTraces(nextTracesConsumer, cfg)

	return processorhelper.NewTraces(
		ctx,
		set,
		cfg,
		nextTracesConsumer,
		st.processTraces,
		processorhelper.WithCapabilities(st.Capabilities()),
		processorhelper.WithStart(st.Start),
		processorhelper.WithShutdown(st.Shutdown))
}

func createMetricsProcessor(
	ctx context.Context,
	set processor.Settings,
	cfg component.Config,
	nextMetricsConsumer consumer.Metrics,
) (processor.Metrics, error) {
	log.Println("Creating Metrics Processor")
	sm := newSlurmInfoMetrics(nextMetricsConsumer, cfg)

	return processorhelper.NewMetrics(
		ctx,
		set,
		cfg,
		nextMetricsConsumer,
		sm.processMetrics,
		processorhelper.WithCapabilities(sm.Capabilities()),
		processorhelper.WithStart(sm.Start),
		processorhelper.WithShutdown(sm.Shutdown))
}

func createLogsProcessor(
	ctx context.Context,
	set processor.Settings,
	cfg component.Config,
	nextLogsConsumer consumer.Logs,
) (processor.Logs, error) {
	log.Println("Creating Logs Processor")
	sm := newSlurmInfoLogs(nextLogsConsumer, cfg)

	return processorhelper.NewLogs(
		ctx,
		set,
		cfg,
		nextLogsConsumer,
		sm.processLogs,
		processorhelper.WithCapabilities(sm.Capabilities()),
		processorhelper.WithStart(sm.Start),
		processorhelper.WithShutdown(sm.Shutdown))
}