Skip to content

feat: chlogstore supports multi-deployment settings #577

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
alexluong wants to merge 7 commits into
base: clickhouse
Choose a base branch
from
Draft

feat: chlogstore supports multi-deployment settings #577

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
ba0f6a9
chore: include deployment_id in logstore init
alexluong Dec 2, 2025
dac03a7
feat: chlogstore support deployment_id via table
alexluong Dec 2, 2025
9d1e5e8
feat: clickhouse migrator support deployment_id
alexluong Dec 2, 2025
8094648
test: update e2e deployment_id test suite to use chlogstore instead o…
alexluong Dec 2, 2025
d92b099
test: e2e test using logstore
alexluong Dec 2, 2025
aeb8190
fix: migrator config missing deployment_id
alexluong Dec 2, 2025
381fff2
refactor: prefix table instead of suffix
alexluong Dec 2, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
292 changes: 292 additions & 0 deletions cmd/e2e/api_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1861,3 +1861,295 @@ func makeDestinationDisabledValidator(id string, disabled bool) map[string]any {
},
}
}

func (suite *basicSuite) TestLogStoreAPI() {
tenantID := idgen.String()
destinationID := idgen.Destination()
eventID := idgen.Event()

tests := []APITest{
// Setup: Create tenant
{
Name: "PUT /:tenantID - Create tenant",
Request: suite.AuthRequest(httpclient.Request{
Method: httpclient.MethodPUT,
Path: "/" + tenantID,
}),
Expected: APITestExpectation{
Match: &httpclient.Response{
StatusCode: http.StatusCreated,
},
},
},
// Setup: Configure mock server destination
{
Name: "PUT mockserver/destinations",
Request: httpclient.Request{
Method: httpclient.MethodPUT,
BaseURL: suite.mockServerBaseURL,
Path: "/destinations",
Body: map[string]interface{}{
"id": destinationID,
"type": "webhook",
"config": map[string]interface{}{
"url": fmt.Sprintf("%s/webhook/%s", suite.mockServerBaseURL, destinationID),
},
},
},
Expected: APITestExpectation{
Match: &httpclient.Response{
StatusCode: http.StatusOK,
},
},
},
// Setup: Create destination in Outpost
{
Name: "POST /:tenantID/destinations",
Request: suite.AuthRequest(httpclient.Request{
Method: httpclient.MethodPOST,
Path: "/" + tenantID + "/destinations",
Body: map[string]interface{}{
"id": destinationID,
"type": "webhook",
"topics": "*",
"config": map[string]interface{}{
"url": fmt.Sprintf("%s/webhook/%s", suite.mockServerBaseURL, destinationID),
},
},
}),
Expected: APITestExpectation{
Match: &httpclient.Response{
StatusCode: http.StatusCreated,
},
},
},
// Publish an event
{
Name: "POST /publish - Publish event",
Request: suite.AuthRequest(httpclient.Request{
Method: httpclient.MethodPOST,
Path: "/publish",
Body: map[string]interface{}{
"tenant_id": tenantID,
"topic": "user.created",
"eligible_for_retry": true,
"id": eventID,
"metadata": map[string]any{
"source": "test",
},
"data": map[string]any{
"user_id": "123",
"email": "[email protected]",
},
},
}),
Expected: APITestExpectation{
Match: &httpclient.Response{
StatusCode: http.StatusAccepted,
},
},
},
// Wait for event to be processed and logged, then list events
{
Name: "GET /:tenantID/events - List events (with delay)",
Delay: 3 * time.Second,
Request: suite.AuthRequest(httpclient.Request{
Method: httpclient.MethodGET,
Path: "/" + tenantID + "/events",
}),
Expected: APITestExpectation{
Match: &httpclient.Response{
StatusCode: http.StatusOK,
},
Validate: map[string]interface{}{
"type": "object",
"properties": map[string]interface{}{
"statusCode": map[string]interface{}{
"type": "integer",
"const": 200,
},
"body": map[string]interface{}{
"type": "object",
"properties": map[string]interface{}{
"data": map[string]interface{}{
"type": "array",
"minItems": 1,
},
"count": map[string]interface{}{
"type": "integer",
"minimum": 1,
},
},
"required": []any{"data", "count"},
},
},
"required": []any{"statusCode", "body"},
},
},
},
// Retrieve specific event
{
Name: "GET /:tenantID/events/:eventID - Retrieve event",
Request: suite.AuthRequest(httpclient.Request{
Method: httpclient.MethodGET,
Path: "/" + tenantID + "/events/" + eventID,
}),
Expected: APITestExpectation{
Match: &httpclient.Response{
StatusCode: http.StatusOK,
},
Validate: map[string]interface{}{
"type": "object",
"properties": map[string]interface{}{
"statusCode": map[string]interface{}{
"type": "integer",
"const": 200,
},
"body": map[string]interface{}{
"type": "object",
"properties": map[string]interface{}{
"id": map[string]interface{}{
"type": "string",
"const": eventID,
},
"tenant_id": map[string]interface{}{
"type": "string",
"const": tenantID,
},
"topic": map[string]interface{}{
"type": "string",
"const": "user.created",
},
"status": map[string]interface{}{
"type": "string",
"enum": []any{"pending", "success", "failed"},
},
},
"required": []any{"id", "tenant_id", "topic"},
},
},
"required": []any{"statusCode", "body"},
},
},
},
// List deliveries for the event
{
Name: "GET /:tenantID/events/:eventID/deliveries - List deliveries",
Request: suite.AuthRequest(httpclient.Request{
Method: httpclient.MethodGET,
Path: "/" + tenantID + "/events/" + eventID + "/deliveries",
}),
Expected: APITestExpectation{
Match: &httpclient.Response{
StatusCode: http.StatusOK,
},
Validate: map[string]interface{}{
"type": "object",
"properties": map[string]interface{}{
"statusCode": map[string]interface{}{
"type": "integer",
"const": 200,
},
"body": map[string]interface{}{
"type": "array",
"minItems": 1,
"items": map[string]interface{}{
"type": "object",
"properties": map[string]interface{}{
"id": map[string]interface{}{
"type": "string",
},
"status": map[string]interface{}{
"type": "string",
"enum": []any{"pending", "success", "failed"},
},
"delivered_at": map[string]interface{}{
"type": "string",
},
},
"required": []any{"id", "status", "delivered_at"},
},
},
},
"required": []any{"statusCode", "body"},
},
},
},
// List events by destination
{
Name: "GET /:tenantID/destinations/:destinationID/events - List events by destination",
Request: suite.AuthRequest(httpclient.Request{
Method: httpclient.MethodGET,
Path: "/" + tenantID + "/destinations/" + destinationID + "/events",
}),
Expected: APITestExpectation{
Match: &httpclient.Response{
StatusCode: http.StatusOK,
},
Validate: map[string]interface{}{
"type": "object",
"properties": map[string]interface{}{
"statusCode": map[string]interface{}{
"type": "integer",
"const": 200,
},
"body": map[string]interface{}{
"type": "object",
"properties": map[string]interface{}{
"data": map[string]interface{}{
"type": "array",
"minItems": 1,
},
"count": map[string]interface{}{
"type": "integer",
"minimum": 1,
},
},
"required": []any{"data", "count"},
},
},
"required": []any{"statusCode", "body"},
},
},
},
// Retrieve event by destination
{
Name: "GET /:tenantID/destinations/:destinationID/events/:eventID - Retrieve event by destination",
Request: suite.AuthRequest(httpclient.Request{
Method: httpclient.MethodGET,
Path: "/" + tenantID + "/destinations/" + destinationID + "/events/" + eventID,
}),
Expected: APITestExpectation{
Match: &httpclient.Response{
StatusCode: http.StatusOK,
},
Validate: map[string]interface{}{
"type": "object",
"properties": map[string]interface{}{
"statusCode": map[string]interface{}{
"type": "integer",
"const": 200,
},
"body": map[string]interface{}{
"type": "object",
"properties": map[string]interface{}{
"id": map[string]interface{}{
"type": "string",
"const": eventID,
},
"destination_id": map[string]interface{}{
"type": "string",
"const": destinationID,
},
},
"required": []any{"id", "destination_id"},
},
},
"required": []any{"statusCode", "body"},
},
},
},
}

suite.RunAPITests(suite.T(), tests)
}
2 changes: 1 addition & 1 deletion cmd/e2e/suites_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -219,7 +219,7 @@ func TestBasicSuiteWithDeploymentID(t *testing.T) {
}

suite.Run(t, &basicSuite{
logStorageType: configs.LogStorageTypePostgres,
logStorageType: configs.LogStorageTypeClickHouse,
deploymentID: "dp_e2e_test",
})
}
9 changes: 5 additions & 4 deletions internal/config/config.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -473,10 +473,11 @@ func (c *Config) ToMigratorOpts() migrator.MigrationOpts {
URL: c.PostgresURL,
},
CH: migrator.MigrationOptsCH{
Addr: c.ClickHouse.Addr,
Username: c.ClickHouse.Username,
Password: c.ClickHouse.Password,
Database: c.ClickHouse.Database,
Addr: c.ClickHouse.Addr,
Username: c.ClickHouse.Username,
Password: c.ClickHouse.Password,
Database: c.ClickHouse.Database,
DeploymentID: c.DeploymentID,
},
}
}
Loading
Loading