Sidecar: fix startup sequence

Previously we defered starting the http servers by blocking the whole startup until we could ping prometheus. This breaks usecases that rely on the config reloader to start prometheus. We fix it by using a channel to defer starting the http/grpc servers and loading external labels in an actor concurrently. Signed-off-by: Michael Hoffmann <mhoffm@posteo.de>
thanos-io · May 30, 2024 · 20a8f3e · 20a8f3e
1 parent 863d914
commit 20a8f3e
Show file tree

Hide file tree

Showing 2 changed files with 97 additions and 73 deletions.
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -14,6 +14,7 @@ We use *breaking :warning:* to mark changes that are not backward compatible (re
 
 - [#7326](https://github.com/thanos-io/thanos/pull/7326) Query: fixing exemplars proxy when querying stores with multiple tenants.
 - [#7335](https://github.com/thanos-io/thanos/pull/7335) Dependency: Update minio-go to v7.0.70 which includes support for EKS Pod Identity.
+- [#7403](https://github.com/thanos-io/thanos/pull/7403) Sidecar: fix startup sequence
 
 ### Added
 

diff --git a/cmd/thanos/sidecar.go b/cmd/thanos/sidecar.go
@@ -149,22 +149,9 @@ func runSidecar(
 		prober.NewInstrumentation(comp, logger, extprom.WrapRegistererWithPrefix("thanos_", reg)),
 	)
 
-	srv := httpserver.New(logger, reg, comp, httpProbe,
-		httpserver.WithListen(conf.http.bindAddress),
-		httpserver.WithGracePeriod(time.Duration(conf.http.gracePeriod)),
-		httpserver.WithTLSConfig(conf.http.tlsConfig),
-	)
-
-	g.Add(func() error {
-		statusProber.Healthy()
-
-		return srv.ListenAndServe()
-	}, func(err error) {
-		statusProber.NotReady(err)
-		defer statusProber.NotHealthy(err)
-
-		srv.Shutdown(err)
-	})
+	// Once we have loaded external labels from prometheus we can use this to signal the servers
+	// that they can start now.
+	readyToStartServers := make(chan struct{})
 
 	// Setup all the concurrent groups.
 	{
@@ -173,87 +160,88 @@ func runSidecar(
 			Help: "Boolean indicator whether the sidecar can reach its Prometheus peer.",
 		})
 
-		ctx := context.Background()
-		// Only check Prometheus's flags when upload is enabled.
-		if uploads {
-			// Check prometheus's flags to ensure same sidecar flags.
-			// We retry infinitely until we validated prometheus flags
+		ctx, cancel := context.WithCancel(context.Background())
+		g.Add(func() error {
+			// Only check Prometheus's flags when upload is enabled.
+			if uploads {
+				// Check prometheus's flags to ensure same sidecar flags.
+				// We retry infinitely until we validated prometheus flags
+				err := runutil.Retry(conf.prometheus.getConfigInterval, ctx.Done(), func() error {
+					iterCtx, iterCancel := context.WithTimeout(context.Background(), conf.prometheus.getConfigTimeout)
+					defer iterCancel()
+
+					if err := validatePrometheus(iterCtx, m.client, logger, conf.shipper.ignoreBlockSize, m); err != nil {
+						level.Warn(logger).Log(
+							"msg", "failed to validate prometheus flags. Is Prometheus running? Retrying",
+							"err", err,
+						)
+						return err
+					}
+
+					level.Info(logger).Log(
+						"msg", "successfully validated prometheus flags",
+					)
+					return nil
+				})
+				if err != nil {
+					return errors.Wrap(err, "failed to validate prometheus flags")
+				}
+			}
+
+			// We retry infinitely until we reach and fetch BuildVersion from our Prometheus.
 			err := runutil.Retry(conf.prometheus.getConfigInterval, ctx.Done(), func() error {
 				iterCtx, iterCancel := context.WithTimeout(context.Background(), conf.prometheus.getConfigTimeout)
 				defer iterCancel()
 
-				if err := validatePrometheus(iterCtx, m.client, logger, conf.shipper.ignoreBlockSize, m); err != nil {
+				if err := m.BuildVersion(iterCtx); err != nil {
 					level.Warn(logger).Log(
-						"msg", "failed to validate prometheus flags. Is Prometheus running? Retrying",
+						"msg", "failed to fetch prometheus version. Is Prometheus running? Retrying",
 						"err", err,
 					)
 					return err
 				}
 
 				level.Info(logger).Log(
-					"msg", "successfully validated prometheus flags",
+					"msg", "successfully loaded prometheus version",
 				)
 				return nil
 			})
 			if err != nil {
-				return errors.Wrap(err, "failed to validate prometheus flags")
-			}
-		}
-
-		// We retry infinitely until we reach and fetch BuildVersion from our Prometheus.
-		err := runutil.Retry(conf.prometheus.getConfigInterval, ctx.Done(), func() error {
-			iterCtx, iterCancel := context.WithTimeout(context.Background(), conf.prometheus.getConfigTimeout)
-			defer iterCancel()
-
-			if err := m.BuildVersion(iterCtx); err != nil {
-				level.Warn(logger).Log(
-					"msg", "failed to fetch prometheus version. Is Prometheus running? Retrying",
-					"err", err,
-				)
-				return err
+				return errors.Wrap(err, "failed to get prometheus version")
 			}
 
-			level.Info(logger).Log(
-				"msg", "successfully loaded prometheus version",
-			)
-			return nil
-		})
-		if err != nil {
-			return errors.Wrap(err, "failed to get prometheus version")
-		}
+			// Blocking query of external labels before joining as a Source Peer into gossip.
+			// We retry infinitely until we reach and fetch labels from our Prometheus.
+			err = runutil.Retry(conf.prometheus.getConfigInterval, ctx.Done(), func() error {
+				iterCtx, iterCancel := context.WithTimeout(context.Background(), conf.prometheus.getConfigTimeout)
+				defer iterCancel()
 
-		// Blocking query of external labels before joining as a Source Peer into gossip.
-		// We retry infinitely until we reach and fetch labels from our Prometheus.
-		err = runutil.Retry(conf.prometheus.getConfigInterval, ctx.Done(), func() error {
-			iterCtx, iterCancel := context.WithTimeout(context.Background(), conf.prometheus.getConfigTimeout)
-			defer iterCancel()
+				if err := m.UpdateLabels(iterCtx); err != nil {
+					level.Warn(logger).Log(
+						"msg", "failed to fetch initial external labels. Is Prometheus running? Retrying",
+						"err", err,
+					)
+					return err
+				}
 
-			if err := m.UpdateLabels(iterCtx); err != nil {
-				level.Warn(logger).Log(
-					"msg", "failed to fetch initial external labels. Is Prometheus running? Retrying",
-					"err", err,
+				level.Info(logger).Log(
+					"msg", "successfully loaded prometheus external labels",
+					"external_labels", m.Labels().String(),
 				)
-				return err
+				return nil
+			})
+			if err != nil {
+				return errors.Wrap(err, "initial external labels query")
 			}
 
-			level.Info(logger).Log(
-				"msg", "successfully loaded prometheus external labels",
-				"external_labels", m.Labels().String(),
-			)
-			return nil
-		})
-		if err != nil {
-			return errors.Wrap(err, "initial external labels query")
-		}
+			if len(m.Labels()) == 0 {
+				return errors.New("no external labels configured on Prometheus server, uniquely identifying external labels must be configured; see https://thanos.io/tip/thanos/storage.md#external-labels for details.")
+			}
+			promUp.Set(1)
+			statusProber.Ready()
 
-		if len(m.Labels()) == 0 {
-			return errors.New("no external labels configured on Prometheus server, uniquely identifying external labels must be configured; see https://thanos.io/tip/thanos/storage.md#external-labels for details.")
-		}
-		promUp.Set(1)
-		statusProber.Ready()
+			close(readyToStartServers)
 
-		ctx, cancel := context.WithCancel(context.Background())
-		g.Add(func() error {
 			// Periodically query the Prometheus config. We use this as a heartbeat as well as for updating
 			// the external labels we apply.
 			return runutil.Repeat(conf.prometheus.getConfigInterval, ctx.Done(), func() error {
@@ -336,14 +324,49 @@ func runSidecar(
 			grpcserver.WithMaxConnAge(conf.grpc.maxConnectionAge),
 			grpcserver.WithTLSConfig(tlsCfg),
 		)
+
+		ctx, cancel := context.WithCancel(context.Background())
 		g.Add(func() error {
+			select {
+			case <-ctx.Done():
+				return ctx.Err()
+			case <-readyToStartServers:
+			}
+
 			statusProber.Ready()
 			return s.ListenAndServe()
 		}, func(err error) {
+			cancel()
 			statusProber.NotReady(err)
 			s.Shutdown(err)
 		})
 	}
+	{
+		srv := httpserver.New(logger, reg, comp, httpProbe,
+			httpserver.WithListen(conf.http.bindAddress),
+			httpserver.WithGracePeriod(time.Duration(conf.http.gracePeriod)),
+			httpserver.WithTLSConfig(conf.http.tlsConfig),
+		)
+
+		ctx, cancel := context.WithCancel(context.Background())
+		g.Add(func() error {
+			select {
+			case <-ctx.Done():
+				return ctx.Err()
+			case <-readyToStartServers:
+			}
+
+			statusProber.Healthy()
+			return srv.ListenAndServe()
+		}, func(err error) {
+			cancel()
+
+			statusProber.NotReady(err)
+			defer statusProber.NotHealthy(err)
+
+			srv.Shutdown(err)
+		})
+	}
 
 	if uploads {
 		// The background shipper continuously scans the data directory and uploads