Skip to content

Commit 9cd7bcb

Browse files
ash2kmdelapenya
ash2k
and
mdelapenya
authored
Log retried errors (#2613)
Co-authored-by: Manuel de la Peña <[email protected]>
1 parent d60fc7c commit 9cd7bcb

File tree

4 files changed

+172
-157
lines changed

4 files changed

+172
-157
lines changed

‎docker.go

+57-41
Original file line numberDiff line numberDiff line change
@@ -889,20 +889,25 @@ func (p *DockerProvider) BuildImage(ctx context.Context, img ImageBuildInfo) (st
889889

890890
var buildError error
891891
var resp types.ImageBuildResponse
892-
err = backoff.Retry(func() error {
893-
resp, err = p.client.ImageBuild(ctx, buildOptions.Context, buildOptions)
894-
if err != nil {
895-
buildError = errors.Join(buildError, err)
896-
if isPermanentClientError(err) {
897-
return backoff.Permanent(err)
892+
err = backoff.RetryNotify(
893+
func() error {
894+
resp, err = p.client.ImageBuild(ctx, buildOptions.Context, buildOptions)
895+
if err != nil {
896+
buildError = errors.Join(buildError, err)
897+
if isPermanentClientError(err) {
898+
return backoff.Permanent(err)
899+
}
900+
return err
898901
}
899-
Logger.Printf("Failed to build image: %s, will retry", err)
900-
return err
901-
}
902-
defer p.Close()
903-
904-
return nil
905-
}, backoff.WithContext(backoff.NewExponentialBackOff(), ctx))
902+
defer p.Close()
903+
904+
return nil
905+
},
906+
backoff.WithContext(backoff.NewExponentialBackOff(), ctx),
907+
func(err error, duration time.Duration) {
908+
p.Logger.Printf("Failed to build image: %s, will retry", err)
909+
},
910+
)
906911
if err != nil {
907912
return "", errors.Join(buildError, err)
908913
}
@@ -1015,7 +1020,7 @@ func (p *DockerProvider) CreateContainer(ctx context.Context, req ContainerReque
10151020
}
10161021

10171022
if modifiedTag != imageName {
1018-
Logger.Printf("✍🏼 Replacing image with %s. From: %s to %s\n", is.Description(), imageName, modifiedTag)
1023+
p.Logger.Printf("✍🏼 Replacing image with %s. From: %s to %s\n", is.Description(), imageName, modifiedTag)
10191024
imageName = modifiedTag
10201025
}
10211026
}
@@ -1180,23 +1185,29 @@ func (p *DockerProvider) findContainerByName(ctx context.Context, name string) (
11801185
}
11811186

11821187
func (p *DockerProvider) waitContainerCreation(ctx context.Context, name string) (*types.Container, error) {
1183-
var ctr *types.Container
1184-
return ctr, backoff.Retry(func() error {
1185-
c, err := p.findContainerByName(ctx, name)
1186-
if err != nil {
1187-
if !errdefs.IsNotFound(err) && isPermanentClientError(err) {
1188-
return backoff.Permanent(err)
1188+
return backoff.RetryNotifyWithData(
1189+
func() (*types.Container, error) {
1190+
c, err := p.findContainerByName(ctx, name)
1191+
if err != nil {
1192+
if !errdefs.IsNotFound(err) && isPermanentClientError(err) {
1193+
return nil, backoff.Permanent(err)
1194+
}
1195+
return nil, err
11891196
}
1190-
return err
1191-
}
11921197

1193-
if c == nil {
1194-
return fmt.Errorf("container %s not found", name)
1195-
}
1196-
1197-
ctr = c
1198-
return nil
1199-
}, backoff.WithContext(backoff.NewExponentialBackOff(), ctx))
1198+
if c == nil {
1199+
return nil, errdefs.NotFound(fmt.Errorf("container %s not found", name))
1200+
}
1201+
return c, nil
1202+
},
1203+
backoff.WithContext(backoff.NewExponentialBackOff(), ctx),
1204+
func(err error, duration time.Duration) {
1205+
if errdefs.IsNotFound(err) {
1206+
return
1207+
}
1208+
p.Logger.Printf("Waiting for container. Got an error: %v; Retrying in %d seconds", err, duration/time.Second)
1209+
},
1210+
)
12001211
}
12011212

12021213
func (p *DockerProvider) ReuseOrCreateContainer(ctx context.Context, req ContainerRequest) (Container, error) {
@@ -1283,19 +1294,24 @@ func (p *DockerProvider) attemptToPullImage(ctx context.Context, tag string, pul
12831294
}
12841295

12851296
var pull io.ReadCloser
1286-
err = backoff.Retry(func() error {
1287-
pull, err = p.client.ImagePull(ctx, tag, pullOpt)
1288-
if err != nil {
1289-
if isPermanentClientError(err) {
1290-
return backoff.Permanent(err)
1297+
err = backoff.RetryNotify(
1298+
func() error {
1299+
pull, err = p.client.ImagePull(ctx, tag, pullOpt)
1300+
if err != nil {
1301+
if isPermanentClientError(err) {
1302+
return backoff.Permanent(err)
1303+
}
1304+
return err
12911305
}
1292-
Logger.Printf("Failed to pull image: %s, will retry", err)
1293-
return err
1294-
}
1295-
defer p.Close()
1296-
1297-
return nil
1298-
}, backoff.WithContext(backoff.NewExponentialBackOff(), ctx))
1306+
defer p.Close()
1307+
1308+
return nil
1309+
},
1310+
backoff.WithContext(backoff.NewExponentialBackOff(), ctx),
1311+
func(err error, duration time.Duration) {
1312+
p.Logger.Printf("Failed to pull image: %s, will retry", err)
1313+
},
1314+
)
12991315
if err != nil {
13001316
return err
13011317
}

‎lifecycle.go

+27-21
Original file line numberDiff line numberDiff line change
@@ -215,33 +215,39 @@ var defaultReadinessHook = func() ContainerLifecycleHooks {
215215
b.MaxElapsedTime = 1 * time.Second
216216
b.MaxInterval = 5 * time.Second
217217

218-
err := backoff.Retry(func() error {
219-
jsonRaw, err := dockerContainer.inspectRawContainer(ctx)
220-
if err != nil {
221-
return err
222-
}
218+
err := backoff.RetryNotify(
219+
func() error {
220+
jsonRaw, err := dockerContainer.inspectRawContainer(ctx)
221+
if err != nil {
222+
return err
223+
}
223224

224-
exposedAndMappedPorts := jsonRaw.NetworkSettings.Ports
225-
226-
for _, exposedPort := range dockerContainer.exposedPorts {
227-
portMap := nat.Port(exposedPort)
228-
// having entries in exposedAndMappedPorts, where the key is the exposed port,
229-
// and the value is the mapped port, means that the port has been already mapped.
230-
if _, ok := exposedAndMappedPorts[portMap]; !ok {
231-
// check if the port is mapped with the protocol (default is TCP)
232-
if !strings.Contains(exposedPort, "/") {
233-
portMap = nat.Port(fmt.Sprintf("%s/tcp", exposedPort))
234-
if _, ok := exposedAndMappedPorts[portMap]; !ok {
225+
exposedAndMappedPorts := jsonRaw.NetworkSettings.Ports
226+
227+
for _, exposedPort := range dockerContainer.exposedPorts {
228+
portMap := nat.Port(exposedPort)
229+
// having entries in exposedAndMappedPorts, where the key is the exposed port,
230+
// and the value is the mapped port, means that the port has been already mapped.
231+
if _, ok := exposedAndMappedPorts[portMap]; !ok {
232+
// check if the port is mapped with the protocol (default is TCP)
233+
if !strings.Contains(exposedPort, "/") {
234+
portMap = nat.Port(fmt.Sprintf("%s/tcp", exposedPort))
235+
if _, ok := exposedAndMappedPorts[portMap]; !ok {
236+
return fmt.Errorf("port %s is not mapped yet", exposedPort)
237+
}
238+
} else {
235239
return fmt.Errorf("port %s is not mapped yet", exposedPort)
236240
}
237-
} else {
238-
return fmt.Errorf("port %s is not mapped yet", exposedPort)
239241
}
240242
}
241-
}
242243

243-
return nil
244-
}, b)
244+
return nil
245+
},
246+
b,
247+
func(err error, duration time.Duration) {
248+
dockerContainer.logger.Printf("All requested ports were not exposed: %v", err)
249+
},
250+
)
245251
if err != nil {
246252
return fmt.Errorf("all exposed ports, %s, were not mapped in 5s: %w", dockerContainer.exposedPorts, err)
247253
}

‎modules/clickhouse/clickhouse_test.go

+52-57
Original file line numberDiff line numberDiff line change
@@ -4,6 +4,7 @@ import (
44
"context"
55
"path/filepath"
66
"testing"
7+
"time"
78

89
ch "github.com/ClickHouse/clickhouse-go/v2"
910
"github.com/ClickHouse/clickhouse-go/v2/lib/driver"
@@ -95,7 +96,7 @@ func TestClickHouseConnectionHost(t *testing.T) {
9596
defer conn.Close()
9697

9798
// perform assertions
98-
data, err := performCRUD(conn)
99+
data, err := performCRUD(t, conn)
99100
require.NoError(t, err)
100101
assert.Len(t, data, 1)
101102
}
@@ -132,7 +133,7 @@ func TestClickHouseDSN(t *testing.T) {
132133
defer conn.Close()
133134

134135
// perform assertions
135-
data, err := performCRUD(conn)
136+
data, err := performCRUD(t, conn)
136137
require.NoError(t, err)
137138
assert.Len(t, data, 1)
138139
}
@@ -223,7 +224,7 @@ func TestClickHouseWithConfigFile(t *testing.T) {
223224
defer conn.Close()
224225

225226
// perform assertions
226-
data, err := performCRUD(conn)
227+
data, err := performCRUD(t, conn)
227228
require.NoError(t, err)
228229
assert.Len(t, data, 1)
229230
})
@@ -287,75 +288,69 @@ func TestClickHouseWithZookeeper(t *testing.T) {
287288
defer conn.Close()
288289

289290
// perform assertions
290-
data, err := performReplicatedCRUD(conn)
291+
data, err := performReplicatedCRUD(t, conn)
291292
require.NoError(t, err)
292293
assert.Len(t, data, 1)
293294
}
294295

295-
func performReplicatedCRUD(conn driver.Conn) ([]Test, error) {
296-
var (
297-
err error
298-
res []Test
299-
)
300-
301-
err = backoff.Retry(func() error {
302-
err = conn.Exec(context.Background(), "CREATE TABLE replicated_test_table (id UInt64) ENGINE = ReplicatedMergeTree('/clickhouse/tables/{shard}/mdb.data_transfer_cp_cdc', '{replica}') PRIMARY KEY (id) ORDER BY (id) SETTINGS index_granularity = 8192;")
303-
if err != nil {
304-
return err
305-
}
306-
307-
err = conn.Exec(context.Background(), "INSERT INTO replicated_test_table (id) VALUES (1);")
308-
if err != nil {
309-
return err
310-
}
311-
312-
rows, err := conn.Query(context.Background(), "SELECT * FROM replicated_test_table;")
313-
if err != nil {
314-
return err
315-
}
296+
func performReplicatedCRUD(t *testing.T, conn driver.Conn) ([]Test, error) {
297+
return backoff.RetryNotifyWithData(
298+
func() ([]Test, error) {
299+
err := conn.Exec(context.Background(), "CREATE TABLE replicated_test_table (id UInt64) ENGINE = ReplicatedMergeTree('/clickhouse/tables/{shard}/mdb.data_transfer_cp_cdc', '{replica}') PRIMARY KEY (id) ORDER BY (id) SETTINGS index_granularity = 8192;")
300+
if err != nil {
301+
return nil, err
302+
}
316303

317-
for rows.Next() {
318-
var r Test
304+
err = conn.Exec(context.Background(), "INSERT INTO replicated_test_table (id) VALUES (1);")
305+
if err != nil {
306+
return nil, err
307+
}
319308

320-
err := rows.Scan(&r.Id)
309+
rows, err := conn.Query(context.Background(), "SELECT * FROM replicated_test_table;")
321310
if err != nil {
322-
return err
311+
return nil, err
323312
}
324313

325-
res = append(res, r)
326-
}
327-
return nil
328-
}, backoff.NewExponentialBackOff())
314+
var res []Test
315+
for rows.Next() {
316+
var r Test
329317

330-
return res, err
331-
}
318+
err := rows.Scan(&r.Id)
319+
if err != nil {
320+
return nil, err
321+
}
332322

333-
func performCRUD(conn driver.Conn) ([]Test, error) {
334-
var (
335-
err error
336-
rows []Test
323+
res = append(res, r)
324+
}
325+
return res, nil
326+
},
327+
backoff.NewExponentialBackOff(),
328+
func(err error, duration time.Duration) {
329+
t.Log(err)
330+
},
337331
)
332+
}
338333

339-
err = backoff.Retry(func() error {
340-
err = conn.Exec(context.Background(), "create table if not exists test_table (id UInt64) engine = MergeTree PRIMARY KEY (id) ORDER BY (id) SETTINGS index_granularity = 8192;")
341-
if err != nil {
342-
return err
343-
}
344-
345-
err = conn.Exec(context.Background(), "INSERT INTO test_table (id) VALUES (1);")
346-
if err != nil {
347-
return err
348-
}
349-
350-
rows, err = getAllRows(conn)
351-
if err != nil {
352-
return err
353-
}
334+
func performCRUD(t *testing.T, conn driver.Conn) ([]Test, error) {
335+
return backoff.RetryNotifyWithData(
336+
func() ([]Test, error) {
337+
err := conn.Exec(context.Background(), "create table if not exists test_table (id UInt64) engine = MergeTree PRIMARY KEY (id) ORDER BY (id) SETTINGS index_granularity = 8192;")
338+
if err != nil {
339+
return nil, err
340+
}
354341

355-
return nil
356-
}, backoff.NewExponentialBackOff())
342+
err = conn.Exec(context.Background(), "INSERT INTO test_table (id) VALUES (1);")
343+
if err != nil {
344+
return nil, err
345+
}
357346

358-
return rows, err
347+
return getAllRows(conn)
348+
},
349+
backoff.NewExponentialBackOff(),
350+
func(err error, duration time.Duration) {
351+
t.Log(err)
352+
},
353+
)
359354
}
360355

361356
func getAllRows(conn driver.Conn) ([]Test, error) {

0 commit comments

Comments
 (0)