-
Notifications
You must be signed in to change notification settings - Fork 39
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
[pkg/reciever/jobreceiver] Implement receiver
Implements a feature gated monitoringjob receiver that schedules and executes commands. Signed-off-by: Christian Kruse <[email protected]>
- Loading branch information
Showing
9 changed files
with
263 additions
and
94 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,69 @@ | ||
| package commandtest | ||
|
|
||
| import ( | ||
| "context" | ||
| "flag" | ||
| "fmt" | ||
| "os" | ||
| "os/exec" | ||
| "strconv" | ||
| "strings" | ||
| "testing" | ||
| "time" | ||
| ) | ||
|
|
||
| // WrapTestMain can be used in TestMain to wrap the go test binary with basic | ||
| // command emulation, normalizing the test environment across platforms. | ||
| // | ||
| // Usage: | ||
| // | ||
| // func TestMain(m *testing.M) { | ||
| // commandtest.WrapTestMain(m) | ||
| // } | ||
| func WrapTestMain(m *testing.M) { | ||
| flag.Parse() | ||
|
|
||
| pid := os.Getpid() | ||
| if os.Getenv("GO_EXEC_TEST_PID") == "" { | ||
| os.Setenv("GO_EXEC_TEST_PID", strconv.Itoa(pid)) | ||
| os.Exit(m.Run()) | ||
| } | ||
|
|
||
| args := flag.Args() | ||
| if len(args) == 0 { | ||
| fmt.Fprintf(os.Stderr, "No command\n") | ||
| os.Exit(2) | ||
| } | ||
|
|
||
| command, args := args[0], args[1:] | ||
| switch command { | ||
| case "echo": | ||
| fmt.Fprintf(os.Stdout, "%s", strings.Join(args, " ")) | ||
| case "exit": | ||
| if i, err := strconv.ParseInt(args[0], 10, 32); err == nil { | ||
| os.Exit(int(i)) | ||
| } | ||
| panic("unexpected exit argument") | ||
| case "sleep": | ||
| if d, err := time.ParseDuration(args[0]); err == nil { | ||
| time.Sleep(d) | ||
| return | ||
| } | ||
| if i, err := strconv.ParseInt(args[0], 10, 64); err == nil { | ||
| time.Sleep(time.Second * time.Duration(i)) | ||
| return | ||
| } | ||
| case "fork": | ||
| childCommand := exec.CommandContext(context.Background(), os.Args[0], args...) | ||
| if err := childCommand.Run(); err != nil { | ||
| fmt.Fprintf(os.Stderr, "fork error: %v", err) | ||
| os.Exit(3) | ||
| } | ||
| } | ||
| } | ||
|
|
||
| // WrapCommand adjusts a command and arguments to run emuldated by a go test | ||
| // binary wrapped with WrapTestMain | ||
| func WrapCommand(cmd string, args []string) (string, []string) { | ||
| return os.Args[0], append([]string{cmd}, args...) | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,81 @@ | ||
| package jobreceiver | ||
|
|
||
| import ( | ||
| "context" | ||
| "testing" | ||
| "time" | ||
|
|
||
| "github.com/SumoLogic/sumologic-otel-collector/pkg/receiver/jobreceiver/internal/commandtest" | ||
| "github.com/SumoLogic/sumologic-otel-collector/pkg/receiver/jobreceiver/output" | ||
| "github.com/stretchr/testify/assert" | ||
| "github.com/stretchr/testify/require" | ||
| "go.opentelemetry.io/collector/component/componenttest" | ||
| "go.opentelemetry.io/collector/consumer/consumertest" | ||
| "go.opentelemetry.io/collector/featuregate" | ||
| "go.opentelemetry.io/collector/receiver/receivertest" | ||
| ) | ||
|
|
||
| // TestMain enable command emulation from commandtest | ||
| func TestMain(m *testing.M) { | ||
| commandtest.WrapTestMain(m) | ||
| } | ||
|
|
||
| func TestMonitoringJob(t *testing.T) { | ||
| // basic test | ||
| require.NoError(t, featuregate.GlobalRegistry().Set(featureEnabledId, true)) | ||
| t.Cleanup(func() { require.NoError(t, featuregate.GlobalRegistry().Set(featureEnabledId, false)) }) | ||
|
|
||
| f := NewFactory() | ||
| cfg := testdataConfigSimple() | ||
|
|
||
| sink := new(consumertest.LogsSink) | ||
|
|
||
| rec, err := f.CreateLogsReceiver(context.Background(), receivertest.NewNopCreateSettings(), cfg, sink) | ||
| require.NoError(t, err) | ||
|
|
||
| require.NoError(t, rec.Start(context.Background(), componenttest.NewNopHost())) | ||
|
|
||
| if !assert.Eventually(t, expectNLogs(sink, 1), time.Second*5, time.Millisecond*50, "expected one log entry") { | ||
| t.Fatalf("actual %d, %v", sink.LogRecordCount(), sink.AllLogs()) | ||
| } | ||
| require.NoError(t, rec.Shutdown(context.Background())) | ||
|
|
||
| first := sink.AllLogs()[0] | ||
| firstRecord := first.ResourceLogs().At(0).ScopeLogs().At(0).LogRecords().At(0) | ||
| assert.Equal(t, "hello world", firstRecord.Body().AsString()) | ||
|
|
||
| t.Run("disabled by feature gate", func(t *testing.T) { | ||
| require.NoError(t, featuregate.GlobalRegistry().Set(featureEnabledId, false)) | ||
|
|
||
| f := NewFactory() | ||
| cfg := testdataConfigSimple() | ||
|
|
||
| sink := new(consumertest.LogsSink) | ||
|
|
||
| rec, err := f.CreateLogsReceiver(context.Background(), receivertest.NewNopCreateSettings(), cfg, sink) | ||
| require.NoError(t, err) | ||
|
|
||
| require.NoError(t, rec.Start(context.Background(), componenttest.NewNopHost())) | ||
| // TODO(ck) bleh. | ||
| <-time.After(time.Millisecond * 1500) | ||
|
|
||
| assert.Equal(t, 0, sink.LogRecordCount()) | ||
| }) | ||
| } | ||
|
|
||
| func testdataConfigSimple() *Config { | ||
| cfg := &Config{ | ||
| Exec: newDefaultExecutionConfig(), | ||
| Schedule: ScheduleConfig{Interval: time.Millisecond * 100}, | ||
| Output: output.NewDefaultConfig(), | ||
| } | ||
| cmd, args := commandtest.WrapCommand("echo", []string{"hello world"}) | ||
| cfg.Exec.Command, cfg.Exec.Arguments = cmd, args | ||
| return cfg | ||
| } | ||
|
|
||
| func expectNLogs(sink *consumertest.LogsSink, expected int) func() bool { | ||
| return func() bool { | ||
| return expected <= sink.LogRecordCount() | ||
| } | ||
| } |
Oops, something went wrong.