Skip to content
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.

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

fix(influxdb): Respect custom waitStrategy #2845

Merged
merged 11 commits into from
Nov 20, 2024
66 changes: 35 additions & 31 deletions modules/influxdb/influxdb.go
Original file line number Diff line number Diff line change
Expand Up @@ -2,7 +2,9 @@ package influxdb

import (
"context"
"encoding/json"
"fmt"
"io"
"path"
"strings"

Expand Down Expand Up @@ -34,7 +36,6 @@ func Run(ctx context.Context, img string, opts ...testcontainers.ContainerCustom
"INFLUXDB_HTTP_HTTPS_ENABLED": "false",
"INFLUXDB_HTTP_AUTH_ENABLED": "false",
},
WaitingFor: wait.ForListeningPort("8086/tcp"),
}
genericContainerReq := testcontainers.GenericContainerRequest{
ContainerRequest: req,
Expand All @@ -47,36 +48,8 @@ func Run(ctx context.Context, img string, opts ...testcontainers.ContainerCustom
}
}

hasInitDb := false

for _, f := range genericContainerReq.Files {
if f.ContainerFilePath == "/" && strings.HasSuffix(f.HostFilePath, "docker-entrypoint-initdb.d") {
// Init service in container will start influxdb, run scripts in docker-entrypoint-initdb.d and then
// terminate the influxdb server, followed by restart of influxdb. This is tricky to wait for, and
// in this case, we are assuming that data was added by init script, so we then look for an
// "Open shard" which is the last thing that happens before the server is ready to accept connections.
// This is probably different for InfluxDB 2.x, but that is left as an exercise for the reader.
strategies := []wait.Strategy{
genericContainerReq.WaitingFor,
wait.ForLog("influxdb init process in progress..."),
wait.ForLog("Server shutdown completed"),
wait.ForLog("Opened shard"),
}
genericContainerReq.WaitingFor = wait.ForAll(strategies...)
hasInitDb = true
break
}
}

if !hasInitDb {
if lastIndex := strings.LastIndex(genericContainerReq.Image, ":"); lastIndex != -1 {
tag := genericContainerReq.Image[lastIndex+1:]
if tag == "latest" || tag[0] == '2' {
genericContainerReq.WaitingFor = wait.ForLog(`Listening log_id=[0-9a-zA-Z_~]+ service=tcp-listener transport=http`).AsRegexp()
}
} else {
genericContainerReq.WaitingFor = wait.ForLog("Listening for signals")
}
if genericContainerReq.WaitingFor == nil {
genericContainerReq.WaitingFor = defaultWaitStrategy(genericContainerReq)
}

container, err := testcontainers.GenericContainer(ctx, genericContainerReq)
Expand All @@ -92,6 +65,37 @@ func Run(ctx context.Context, img string, opts ...testcontainers.ContainerCustom
return c, nil
}

func defaultWaitStrategy(genericContainerReq testcontainers.GenericContainerRequest) wait.Strategy {
for _, f := range genericContainerReq.Files {
if f.ContainerFilePath == "/" && strings.HasSuffix(f.HostFilePath, "docker-entrypoint-initdb.d") {
// Init service in container will start influxdb, run scripts in docker-entrypoint-initdb.d and then
// terminate the influxdb server, followed by restart of influxdb. This is tricky to wait for, and
// in this case, we are assuming that data was added by init script
// This is probably different for InfluxDB 2.x, but that is left as an exercise for the reader.
strategies := []wait.Strategy{
wait.ForLog("Server shutdown completed"),
waitForHttpHealth(),
}
return wait.ForAll(strategies...)
marcinmilewski93 marked this conversation as resolved.
Show resolved Hide resolved
}
}
return waitForHttpHealth()
}

func waitForHttpHealth() *wait.HTTPStrategy {
return wait.ForHTTP("/health").
mdelapenya marked this conversation as resolved.
Show resolved Hide resolved
WithResponseMatcher(func(body io.Reader) bool {
marcinmilewski93 marked this conversation as resolved.
Show resolved Hide resolved
decoder := json.NewDecoder(body)
r := struct {
Status string `json:"status"`
}{}
if err := decoder.Decode(&r); err != nil {
return false
}
return r.Status == "pass"
})
}

func (c *InfluxDbContainer) MustConnectionUrl(ctx context.Context) string {
connectionString, err := c.ConnectionUrl(ctx)
if err != nil {
Expand Down
Loading