-
Notifications
You must be signed in to change notification settings - Fork 73
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
feat: agent collector statistics report (#3544)
* feat: count spans * agent: implement grpc stream client * agent: implement send otlp connection result * feat: send collector statistics to server when requested * test: check if collector statistics are precise * feat: reset statistics if flag is present * fix issue with datastore connection test and add tests
- Loading branch information
1 parent
eac8b6d
commit a53a0d5
Showing
16 changed files
with
1,214 additions
and
525 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
42 changes: 42 additions & 0 deletions
42
agent/client/workflow_listen_for_ds_connection_tests_test.go
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,42 @@ | ||
package client_test | ||
|
||
import ( | ||
"context" | ||
"testing" | ||
"time" | ||
|
||
"github.com/kubeshop/tracetest/agent/client" | ||
"github.com/kubeshop/tracetest/agent/client/mocks" | ||
"github.com/kubeshop/tracetest/agent/proto" | ||
"github.com/stretchr/testify/assert" | ||
"github.com/stretchr/testify/require" | ||
) | ||
|
||
func TestDataStoreConnectionTestWorkflow(t *testing.T) { | ||
server := mocks.NewGrpcServer() | ||
defer server.Stop() | ||
|
||
client, err := client.Connect(context.Background(), server.Addr()) | ||
require.NoError(t, err) | ||
|
||
var receivedConnectionTestRequest *proto.DataStoreConnectionTestRequest | ||
client.OnDataStoreTestConnectionRequest(func(ctx context.Context, otr *proto.DataStoreConnectionTestRequest) error { | ||
receivedConnectionTestRequest = otr | ||
return nil | ||
}) | ||
|
||
err = client.Start(context.Background()) | ||
require.NoError(t, err) | ||
|
||
connectionTestRequest := &proto.DataStoreConnectionTestRequest{ | ||
RequestID: "request-id", | ||
} | ||
|
||
server.SendDataStoreConnectionTestRequest(connectionTestRequest) | ||
|
||
// ensures there's enough time for networking between server and client | ||
time.Sleep(1 * time.Second) | ||
|
||
assert.NotNil(t, receivedConnectionTestRequest) | ||
assert.Equal(t, connectionTestRequest.RequestID, "request-id") | ||
} |
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,47 @@ | ||
package client | ||
|
||
import ( | ||
"context" | ||
"fmt" | ||
"log" | ||
"time" | ||
|
||
"github.com/kubeshop/tracetest/agent/proto" | ||
) | ||
|
||
func (c *Client) startOTLPConnectionTestListener(ctx context.Context) error { | ||
client := proto.NewOrchestratorClient(c.conn) | ||
|
||
stream, err := client.RegisterOTLPConnectionTestListener(ctx, c.sessionConfig.AgentIdentification) | ||
if err != nil { | ||
return fmt.Errorf("could not open agent stream: %w", err) | ||
} | ||
|
||
go func() { | ||
for { | ||
req := proto.OTLPConnectionTestRequest{} | ||
err := stream.RecvMsg(&req) | ||
if isEndOfFileError(err) || isCancelledError(err) { | ||
return | ||
} | ||
|
||
reconnected, err := c.handleDisconnectionError(err) | ||
if reconnected { | ||
return | ||
} | ||
|
||
if err != nil { | ||
log.Println("could not get message from otlp connection stream: %w", err) | ||
time.Sleep(1 * time.Second) | ||
continue | ||
} | ||
|
||
// TODO: Get ctx from request | ||
err = c.otlpConnectionTestListener(context.Background(), &req) | ||
if err != nil { | ||
fmt.Println(err.Error()) | ||
} | ||
} | ||
}() | ||
return nil | ||
} |
42 changes: 42 additions & 0 deletions
42
agent/client/workflow_listen_for_otlp_connection_tests_test.go
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,42 @@ | ||
package client_test | ||
|
||
import ( | ||
"context" | ||
"testing" | ||
"time" | ||
|
||
"github.com/kubeshop/tracetest/agent/client" | ||
"github.com/kubeshop/tracetest/agent/client/mocks" | ||
"github.com/kubeshop/tracetest/agent/proto" | ||
"github.com/stretchr/testify/assert" | ||
"github.com/stretchr/testify/require" | ||
) | ||
|
||
func TestOtlpConnectionTestWorkflow(t *testing.T) { | ||
server := mocks.NewGrpcServer() | ||
defer server.Stop() | ||
|
||
client, err := client.Connect(context.Background(), server.Addr()) | ||
require.NoError(t, err) | ||
|
||
var receivedConnectionTestRequest *proto.OTLPConnectionTestRequest | ||
client.OnOTLPConnectionTest(func(ctx context.Context, otr *proto.OTLPConnectionTestRequest) error { | ||
receivedConnectionTestRequest = otr | ||
return nil | ||
}) | ||
|
||
err = client.Start(context.Background()) | ||
require.NoError(t, err) | ||
|
||
connectionTestRequest := &proto.OTLPConnectionTestRequest{ | ||
RequestID: "request-id", | ||
} | ||
|
||
server.SendOTLPConnectionTestRequest(connectionTestRequest) | ||
|
||
// ensures there's enough time for networking between server and client | ||
time.Sleep(1 * time.Second) | ||
|
||
assert.NotNil(t, receivedConnectionTestRequest) | ||
assert.Equal(t, connectionTestRequest.RequestID, "request-id") | ||
} |
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,43 @@ | ||
package client_test | ||
|
||
import ( | ||
"context" | ||
"testing" | ||
|
||
"github.com/kubeshop/tracetest/agent/client" | ||
"github.com/kubeshop/tracetest/agent/client/mocks" | ||
"github.com/kubeshop/tracetest/agent/proto" | ||
"github.com/stretchr/testify/assert" | ||
"github.com/stretchr/testify/require" | ||
) | ||
|
||
func TestDataStoreConnectionResult(t *testing.T) { | ||
server := mocks.NewGrpcServer() | ||
defer server.Stop() | ||
|
||
client, err := client.Connect(context.Background(), server.Addr()) | ||
require.NoError(t, err) | ||
|
||
err = client.Start(context.Background()) | ||
require.NoError(t, err) | ||
|
||
result := &proto.DataStoreConnectionTestResponse{ | ||
RequestID: "request-id", | ||
AgentIdentification: &proto.AgentIdentification{}, | ||
Successful: true, | ||
Steps: &proto.DataStoreConnectionTestSteps{ | ||
PortCheck: &proto.DataStoreConnectionTestStep{ | ||
Passed: true, | ||
}, | ||
}, | ||
} | ||
|
||
err = client.SendDataStoreConnectionResult(context.Background(), result) | ||
require.NoError(t, err) | ||
|
||
receivedResponse := server.GetLastDataStoreConnectionResponse() | ||
|
||
assert.Equal(t, result.RequestID, receivedResponse.RequestID) | ||
assert.True(t, result.Successful) | ||
assert.True(t, result.Steps.PortCheck.Passed) | ||
} |
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,21 @@ | ||
package client | ||
|
||
import ( | ||
"context" | ||
"fmt" | ||
|
||
"github.com/kubeshop/tracetest/agent/proto" | ||
) | ||
|
||
func (c *Client) SendOTLPConnectionResult(ctx context.Context, response *proto.OTLPConnectionTestResponse) error { | ||
client := proto.NewOrchestratorClient(c.conn) | ||
|
||
response.AgentIdentification = c.sessionConfig.AgentIdentification | ||
|
||
_, err := client.SendOTLPConnectionTestResult(ctx, response) | ||
if err != nil { | ||
return fmt.Errorf("could not send otlp connection result request: %w", err) | ||
} | ||
|
||
return nil | ||
} |
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,41 @@ | ||
package client_test | ||
|
||
import ( | ||
"context" | ||
"testing" | ||
"time" | ||
|
||
"github.com/kubeshop/tracetest/agent/client" | ||
"github.com/kubeshop/tracetest/agent/client/mocks" | ||
"github.com/kubeshop/tracetest/agent/proto" | ||
"github.com/stretchr/testify/assert" | ||
"github.com/stretchr/testify/require" | ||
) | ||
|
||
func TestOTLPConnectionResultTrace(t *testing.T) { | ||
server := mocks.NewGrpcServer() | ||
defer server.Stop() | ||
|
||
client, err := client.Connect(context.Background(), server.Addr()) | ||
require.NoError(t, err) | ||
|
||
err = client.Start(context.Background()) | ||
require.NoError(t, err) | ||
|
||
now := time.Now() | ||
result := &proto.OTLPConnectionTestResponse{ | ||
RequestID: "request-id", | ||
AgentIdentification: &proto.AgentIdentification{}, | ||
SpanCount: 10, | ||
LastSpanTimestamp: now.UnixMilli(), | ||
} | ||
|
||
err = client.SendOTLPConnectionResult(context.Background(), result) | ||
require.NoError(t, err) | ||
|
||
receivedResponse := server.GetLastOTLPConnectionResponse() | ||
|
||
assert.Equal(t, result.RequestID, receivedResponse.RequestID) | ||
assert.Equal(t, result.SpanCount, receivedResponse.SpanCount) | ||
assert.Equal(t, result.LastSpanTimestamp, receivedResponse.LastSpanTimestamp) | ||
} |
Oops, something went wrong.