[exporter/clickhouse] Fix database creation (open-telemetry#38829)

<!--Ex. Fixing a bug - Describe the bug and how this fixes the issue.
Ex. Adding a feature - Explain what this achieves.-->
#### Description
In the current version of the exporter, the createDatabase function uses
cfg.buildDSN, which appends the database from the config into the dsn
that is passed into the clickhouse driver.
As a result, the exporter tries to connect to the database before it is
actually created, causing a ClickHouse exception in the start function

This pr couldn't be merged until [the integration test
fix](open-telemetry#38798)

<!--Describe what testing was performed and which tests were added.-->
#### Testing
I've added an integration test that checks whether the database was
successfully created. It fails now but works with the fix.

<!--Please delete paragraphs that you did not use before submitting.-->

---------

Co-authored-by: Antoine Toulme <[email protected]>

Author: 2 people

.chloggen/k.garmanov_FIX-CLICKHOUSEEXPORTER-DATABASE-CREATION.yaml

@@ -0,0 +1,27 @@
+# Use this changelog template to create an entry for release notes.
+
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: bug_fix
+
+# The name of the component, or a single word describing the area of concern, (e.g. filelogreceiver)
+component: clickhouseexporter
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: clickhouseexporter doesn't set the database name in the dsn of the query that creates the database
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [38829]
+
+# (Optional) One or more lines of additional information to render under the primary note.
+# These lines will be padded with 2 spaces and then inserted directly into the document.
+# Use pipe (|) for multiline entries.
+subtext:
+
+# If your change doesn't affect end users or the exported elements of any package,
+# you should instead start your pull request title with [chore] or use the "Skip Changelog" label.
+# Optional: The change log or logs in which this entry should be included.
+# e.g. '[user]' or '[user, api]'
+# Include 'user' if the change is relevant to end users.
+# Include 'api' if there is a change to a library API.
+# Default: '[user]'
+change_logs: []

exporter/clickhouseexporter/Makefile

@@ -2,7 +2,7 @@ include ../../Makefile.Common
 
 local-run-example:
 	cd ../../cmd/otelcontribcol && GOOS=linux go build -o ../../local/otelcontribcol
-	cd example && docker-compose up -d
+	cd example && docker compose up -d
 recreate-otel-collector:
 	cd ../../ && make otelcontribcol
-	cd example && docker-compose up --build otelcollector
+	cd example && docker compose up --build otelcollector

exporter/clickhouseexporter/example/docker-compose.yml

@@ -24,7 +24,8 @@ services:
       - "24224:24224" # fluentforwarder
       - "24224:24224/udp" # fluentforwarder
     depends_on:
-      - clickhouse
+      clickhouse:
+        condition: service_healthy
     networks:
       - otel-clickhouse
 
@@ -33,8 +34,13 @@ services:
     ports:
       - "9000:9000"
       - "8123:8123"
+    healthcheck:
+      test: wget --no-verbose --tries=1 --spider http://localhost:8123/ping || exit 1
     networks:
       - otel-clickhouse
+    environment:
+      - CLICKHOUSE_USER=default
+      - CLICKHOUSE_PASSWORD=default
 
   grafana:
     image: grafana/grafana:latest

exporter/clickhouseexporter/example/otel-collector-config.yml

@@ -37,6 +37,8 @@ exporters:
       initial_interval: 5s
       max_interval: 30s
       max_elapsed_time: 300s
+    username: default
+    password: default
 extensions:
   health_check:
   pprof:

exporter/clickhouseexporter/exporter_logs.go

@@ -212,9 +212,15 @@ func createDatabase(ctx context.Context, cfg *Config) error {
 		return nil
 	}
 
+	// We couldn't set a database in the dsn while creating the database,
+	// otherwise, there would be an exception from clickhouse
+	targetDatabase := cfg.Database
+	cfg.Database = defaultDatabase
+
 	db, err := cfg.buildDB()
+	cfg.Database = targetDatabase
 	if err != nil {
-		return err
+		return fmt.Errorf("can't connect to clickhouse: %w", err)
 	}
 	defer func() {
 		_ = db.Close()

exporter/clickhouseexporter/integration_test.go

@@ -41,38 +41,55 @@ func TestIntegration(t *testing.T) {
 	}
 
 	for _, c := range testCase {
-		t.Run(c.name, func(t *testing.T) {
-			port := randPort()
-			req := testcontainers.ContainerRequest{
-				Image:        c.image,
-				ExposedPorts: []string{fmt.Sprintf("%s:9000", port)},
-				WaitingFor: wait.ForListeningPort("9000").
-					WithStartupTimeout(2 * time.Minute),
-			}
-			c := getContainer(t, req)
-			defer func() {
-				err := c.Terminate(context.Background())
-				require.NoError(t, err)
-			}()
-
-			host, err := c.Host(context.Background())
-			require.NoError(t, err)
-			endpoint := fmt.Sprintf("tcp://%s:%s", host, port)
-
+		// Container creation is slow, so create it once per test
+		endpoint := createTestClickhouseContainer(t, c.image)
+		t.Run(c.name+" exporting", func(t *testing.T) {
 			logExporter := newTestLogsExporter(t, endpoint)
 			verifyExportLog(t, logExporter)
 
 			traceExporter := newTestTracesExporter(t, endpoint)
-			require.NoError(t, err)
 			verifyExporterTrace(t, traceExporter)
 
 			metricExporter := newTestMetricsExporter(t, endpoint)
-			require.NoError(t, err)
 			verifyExporterMetric(t, metricExporter)
 		})
+		t.Run(c.name+" database creation", func(t *testing.T) {
+			// Verify that start function returns no error
+			logExporter := newTestLogsExporter(t, endpoint, func(c *Config) { c.Database = "otel" })
+			// Verify that the table was created
+			verifyLogTable(t, logExporter)
+
+			traceExporter := newTestTracesExporter(t, endpoint, func(c *Config) { c.Database = "otel" })
+			verifyTraceTable(t, traceExporter)
+
+			metricsExporter := newTestMetricsExporter(t, endpoint, func(c *Config) { c.Database = "otel" })
+			verifyMetricTable(t, metricsExporter)
+		})
 	}
 }
 
+// returns endpoint
+func createTestClickhouseContainer(t *testing.T, image string) string {
+	port := randPort()
+	req := testcontainers.ContainerRequest{
+		Image:        image,
+		ExposedPorts: []string{fmt.Sprintf("%s:9000", port)},
+		WaitingFor: wait.ForListeningPort("9000").
+			WithStartupTimeout(2 * time.Minute),
+	}
+	c := getContainer(t, req)
+	t.Cleanup(func() {
+		err := c.Terminate(context.Background())
+		require.NoError(t, err)
+	})
+
+	host, err := c.Host(context.Background())
+	require.NoError(t, err)
+	endpoint := fmt.Sprintf("tcp://%s:%s", host, port)
+
+	return endpoint
+}
+
 func getContainer(t *testing.T, req testcontainers.ContainerRequest) testcontainers.Container {
 	require.NoError(t, req.Validate())
 	container, err := testcontainers.GenericContainer(
@@ -603,6 +620,24 @@ func verifySummaryMetric(t *testing.T, db *sqlx.DB) {
 	require.Equal(t, expectSummary, actualSummary)
 }
 
+func verifyLogTable(t *testing.T, logExporter *logsExporter) {
+	db := sqlx.NewDb(logExporter.client, clickhouseDriverName)
+	_, err := db.Query("select * from otel.otel_logs")
+	require.NoError(t, err)
+}
+
+func verifyTraceTable(t *testing.T, traceExporter *tracesExporter) {
+	db := sqlx.NewDb(traceExporter.client, clickhouseDriverName)
+	_, err := db.Query("select * from otel.otel_traces")
+	require.NoError(t, err)
+}
+
+func verifyMetricTable(t *testing.T, metricExporter *metricsExporter) {
+	db := sqlx.NewDb(metricExporter.client, clickhouseDriverName)
+	_, err := db.Query("select * from otel.otel_metrics_gauge")
+	require.NoError(t, err)
+}
+
 func randPort() string {
 	return strconv.Itoa(rand.IntN(999) + 9000)
 }