Pub/Sub Subscriber health check

[gkatzioura]

The purspose of this PR is to add health check to PubSub subscribers.

This is not in conflict with the health check of the pubsub publisher, and can be used in parallel.

The criteria for the health check of a subscriber is the message backlog and the last processed message.

If the message has been recently processed in a reasonable time threshold, then the subscriber is healthy.
If the backlog of messages is big but the subscriber consumes messages then this is a scalling issue.

If there hasn't been any processing of recent messages but the backlog increases, then the subscriber is unhealthy

Since there can be multiple subscriptions, the subscriptions shall be enabled per subscriber creation.

The method that processes the messages shall be wrapped in order for a method tracker to receive the recently processed message.

Eventually there would be multiple health trackers per subscription

Since PubSubInboundChannelAdapter precedes DefaultSubscriberFactory it will register the tracker first and DefaultSubscriberFactory will avoid adding any health checks.

If DefaultSubscriberFactory is used directly is will add a health check if enabled.

Added a separate PubSubExecutorConfiguration since the health check for subscribers should be autoconfigured before the actual PubSub configuration.

[sonarcloud]

Kudos, SonarCloud Quality Gate passed!   

0 Bugs
0 Vulnerabilities
0 Security Hotspots
21 Code Smells

80.9% Coverage
0.0% Duplication

[elefeint]

@gkatzioura Apologies for the delay. You've put a lot of thought and work into the pull request, and we'd like to treat it as carefully.

We currently have some work in-flight restructuring the Pub/Sub properties to enable per-subscription setting customization (example: #562) that will both conflict, and possibly affect direction of your PR. It would be better to rebase and review it in a couple of weeks, once that work is completed.

One question I had about the addition of monitoring metrics -- how will subscription/num_undelivered_messages differ from the built-in Pub/Sub subscription/num_outstanding_messages metric?

[elefeint]

Another directional question -- GKE forces pod restart when it detects unhealthy indicators. But restarting pods when there is already a scaling issue is counterproductive, as the load won't get any better if the pods are thrashing (we've had issues with the existing global indicator under load -- spring-attic/spring-cloud-gcp#2628). So I am wary of indicating down when messagesOverThreshold is detected.

[gkatzioura]

Hi @elefeint
I understand the transitional phase so will rebase from time to time as you progress.

Will answer your questions.

One question I had about the addition of monitoring metrics -- how will subscription/num_undelivered_messages differ from the built-in Pub/Sub subscription/num_outstanding_messages metric?

num_outstanding_messages messages in flight that have been sent to the subscribers
num_undelivered_messages messages in the backlog not sent yet or probably were not able to get processed

Will explain marking unhealthy based on the backlog. In a scenario of multiple pods/workers it is expected to have lot's of outstanding messages, also if subscribers are closed or stopped pulling messages the num_outstanding_messages will be a small number because they cannot pull, thus they would seem healthy while they are not.
By using num_undelivered_messages the criteria is the backlog in combination of not processing any messages.

GKE forces pod restart when it detects unhealthy indicators. But restarting pods when there is already a scaling issue is counterproductive, as the load won't get any better if the pods are thrashing (we've had issues with the existing global indicator under load -- spring-cloud/spring-cloud-gcp#2628). So I am wary of indicating down when messagesOverThreshold is detected.

If a subscriber consumes messages, regardless of how big the backlog gets (num_undelivered_messages) the health indicator will indicate healthy. A user will probably autoscale based on the backlog, or will optimise the code to increase the throughput. But as long as messages are processed and acknowledged the subscriber serves its purpose thus is healthy.

If the subscriber is not healthy messages cannot be processed or acknowledged. An error or exception probably occurs while processing. On this scenario because the indicator of recently completed processing is not there, we need to evaluate if there are no messages on the backlog or if messages do exist and wait to be processed (thus our subscriber is unhealthy).

Based on this logic the following scenarios happen

• If the subscriber has processed messages recently, then by default is healthy, regardless of the backlog size
• If the subscriber has not processed any messages recently and the backlogs is small/empty then the subscriber is healthy since there are no messages to pull
• If the subscriber has not processed any messages and the backlog is big then the subscriber is unhealthy, because there are indeed messages to pull but something probably fails

[elefeint]

Thanks for your patience! We can start working on this PR -- it's very useful to have this per-subscription healthcheck functionality.

First, could you revert the unrelated whitespace changes? This project still uses Spring's checkstyle for now -- https://github.com/spring-cloud/spring-cloud-gcp/blob/main/CONTRIBUTING.adoc#code-formatting-guidelines

[mpeddada1]

Hey @gkatzioura, thank you for waiting and your contribution! PR #638 is merged so you can rebase this PR now.

[google-cla]

All (the pull request submitter and all commit authors) CLAs are signed, but one or more commits were authored or co-authored by someone other than the pull request submitter.

We need to confirm that all authors are ok with their commits being contributed to this project. Please have them confirm that by leaving a comment that contains only @googlebot I consent. in this pull request.

Note to project maintainer: There may be cases where the author cannot leave a comment, or the comment is not properly detected as consent. In those cases, you can manually confirm consent of the commit author(s), and set the cla label to yes (if enabled on your project).

ℹ️ Googlers: Go here for more info.

[google-cla]

All (the pull request submitter and all commit authors) CLAs are signed, but one or more commits were authored or co-authored by someone other than the pull request submitter.

We need to confirm that all authors are ok with their commits being contributed to this project. Please have them confirm that by leaving a comment that contains only @googlebot I consent. in this pull request.

Note to project maintainer: There may be cases where the author cannot leave a comment, or the comment is not properly detected as consent. In those cases, you can manually confirm consent of the commit author(s), and set the cla label to yes (if enabled on your project).

ℹ️ Googlers: Go here for more info.

[gkatzioura]

Hi @elefeint & @mpeddada1 did rebase so seems ok for a review!

[elefeint]

Another batch of comments. I apologize for the delays -- I will also be out most of next week, which will not help matters.

[elefeint]

This is a nice structure. However, after the rebase, we have two "default" subscribers thread pools/executors -- one defined here that is exclusively used for the health checks, and the globalScheduler that's dynamically registered. We've been trying to deprecate the global scheduler bean that is not dynamic that's being extracted here.

Since our subscriber thread pool management is so complicated now (there is a deprecated global subscriber threadpool bean, and then there is a default subscriber threadpool constructed from properties, plus separate configurations possible for each subscription), I think it would be best to have a dedicated thread pool for Pub/Sub healthchecks. Then it will never interfere with production workloads.

Could you:

• rename the subscriber executor provider bean to make it clear that it's for health checks. It can be shared between publisher and subscriber health checks in the future; no need to differentiate here.
• Fold the new dedicated threadpool/executor beans into PubSubSubscriptionHealthIndicatorAutoConfiguration.java; this way it will be possible to turn it off with the rest of configuration.
• Put the publisher thread pool configuration back. Let's keep the scope of this PR to subscription logic only.
• This class, pleasant to look at as it is, will then go away.

[elefeint]

Let's get those in a subclass, for logical grouping.

[gkatzioura]

Has moved to PubSubConfiguration.Health

[elefeint]

What about keeping the current project ID inside of HealthTrackerRegistry? Then it does not need to be set separately. registerTracker() could b eoverloaded to operate on either an object or a string (which can be either short or fully qualified subscription names).

[gkatzioura]

I see. I will have a default-project-id on the HealthTrackerRegistry, the one derived by default by the other modules GcpProjectIdProvider.
If the user wants to use subscriptions from multiple projects (my everyday case), then the user should provide the fully qualified subscription name.
For example projects/<project_name>/subscriptions/<subscription_name> instead <subscription_name>
If short subscription name is provided the the project-id on the registry shall be used.

[elefeint]

This makes sense.

[gkatzioura]

For the direct usage of SubscriberFactory, should GcpPubSubAutoConfiguration.defaultSubscriberFactory() accept an optional HealthTrackerRegistry?

@elefeint
This is correct, probably was missed on rebase. Just added it

spring-cloud-gcp/spring-cloud-gcp-autoconfigure/src/main/java/com/google/cloud/spring/autoconfigure/pubsub/GcpPubSubAutoConfiguration.java

Lines 205 to 245 in a9f3eb4

	@Qualifier("subscriberExecutorProvider") Optional<ExecutorProvider> executorProvider,
	@Qualifier("subscriberSystemExecutorProvider") ObjectProvider<ExecutorProvider> systemExecutorProvider,
	@Qualifier("subscriberFlowControlSettings") ObjectProvider<FlowControlSettings> flowControlSettings,
	@Qualifier("subscriberApiClock") ObjectProvider<ApiClock> apiClock,
	@Qualifier("subscriberRetrySettings") ObjectProvider<RetrySettings> retrySettings,
	@Qualifier("healthTrackerRegistry") ObjectProvider<HealthTrackerRegistry> healthTrackerRegistry,
	@Qualifier("subscriberTransportChannelProvider") TransportChannelProvider subscriberTransportChannelProvider) {
	DefaultSubscriberFactory factory = new DefaultSubscriberFactory(this.finalProjectIdProvider,
	this.gcpPubSubProperties);
	factory.setThreadPoolTaskSchedulerMap(this.threadPoolTaskSchedulerMap);
	factory.setGlobalScheduler(this.globalScheduler);
	if (executorProvider.isPresent()) {
	logger.warn(
	"The subscriberExecutorProvider bean is being deprecated. Please use application.properties to configure properties");
	factory.setExecutorProvider(executorProvider.get());
	}
	factory.setCredentialsProvider(this.finalCredentialsProvider);
	factory.setHeaderProvider(this.headerProvider);
	factory.setChannelProvider(subscriberTransportChannelProvider);
	systemExecutorProvider.ifAvailable(factory::setSystemExecutorProvider);
	if (flowControlSettings.getIfAvailable() != null) {
	logger.warn(
	"The subscriberFlowControlSettings bean is being deprecated. Please use application.properties to configure properties");
	factory.setFlowControlSettings(flowControlSettings.getIfAvailable());
	}
	apiClock.ifAvailable(factory::setApiClock);
	if (retrySettings.getIfAvailable() != null) {
	logger.warn(
	"The subscriberRetrySettings bean is being deprecated. Please use application.properties to configure properties");
	factory.setSubscriberStubRetrySettings(retrySettings.getIfAvailable());
	}
	if (this.gcpPubSubProperties.getSubscriber().getRetryableCodes() != null) {
	factory.setRetryableCodes(gcpPubSubProperties.getSubscriber().getRetryableCodes());
	}
	healthTrackerRegistry.ifAvailable(factory::setHealthTrackerRegistry);
	return factory;

[elefeint]

I think we are in the final stages here -- I've tested the health check with one of our samples, and it works great. Really nice to look at code, too.

No rush on this round of comments -- I am unavailable until Thursday for review. I apologize for the multiple layers of delays here.

• Add documentation
  • add warning that the health indicator will not behave entirely as expected if Dead Letter Queueing is enabled on the subscription being checked (num_undelivered_messages will drop down by itself after DLQ threshold is reached).
• document the new required (lagThresholad, backlogThreshold) and optional (lookupInterval) properties. Describe units of measurement -- seconds, number of messages.
• Javadoc all public interface methods. HealthTrackerRegistry especially needs javadoc, make sure to describe seconds being the units of measurement for lagThreshold -- we've had users confused about granularity before.
• Replace @since 2.0.5 with @since 2.0.6
• Could you take a look at Sonar's code smell list?
  No problem on the thread pool creation duplication; the logic is straightforward enough. In principle, `GcpPubSubAutoConfiguration.createThreadPoolTaskScheduler(Integer executorThreads, String threadName) could be turned into a reusable utility but it's probably not worth it.

[sonarcloud]

Kudos, SonarCloud Quality Gate passed!   

0 Bugs
0 Vulnerabilities
0 Security Hotspots
0 Code Smells

84.7% Coverage
0.0% Duplication

[elefeint]

Thank you for the very useful healthcheck, and putting up with the lengthy review process. This is now good to go.

[gkatzioura]

Hi @elefeint ! I am grateful for your time and assistance on this!