Publish metrics when producing and consuming messages Expose the count of successes and failures when consuming and producing messages on gcloud-pubsub. Bug: Issue 14408 Change-Id: I1878f82fdf8a578660b480ba6433aed18438bb48
diff --git a/src/main/java/com/googlesource/gerrit/plugins/pubsub/PubSubEventSubscriber.java b/src/main/java/com/googlesource/gerrit/plugins/pubsub/PubSubEventSubscriber.java index b8d8c49..a6058ae 100644 --- a/src/main/java/com/googlesource/gerrit/plugins/pubsub/PubSubEventSubscriber.java +++ b/src/main/java/com/googlesource/gerrit/plugins/pubsub/PubSubEventSubscriber.java
@@ -18,6 +18,7 @@ import com.google.cloud.pubsub.v1.AckReplyConsumer; import com.google.cloud.pubsub.v1.MessageReceiver; import com.google.cloud.pubsub.v1.Subscriber; +import com.google.common.annotations.VisibleForTesting; import com.google.common.flogger.FluentLogger; import com.google.gson.Gson; import com.google.inject.Inject; @@ -37,6 +38,7 @@ private static final FluentLogger logger = FluentLogger.forEnclosingClass(); private final Gson gson; + private final PubSubSubscriberMetrics subscriberMetrics; private final String topic; private final Consumer<EventMessage> messageProcessor; private final SubscriberProvider subscriberProvider; @@ -48,9 +50,11 @@ Gson gson, SubscriberProvider subscriberProvider, PubSubConfiguration config, + PubSubSubscriberMetrics subscriberMetrics, @Assisted String topic, @Assisted Consumer<EventMessage> messageProcessor) { this.gson = gson; + this.subscriberMetrics = subscriberMetrics; this.topic = topic; this.messageProcessor = messageProcessor; this.subscriberProvider = subscriberProvider; @@ -58,15 +62,8 @@ } public void subscribe() { - MessageReceiver receiver = - (PubsubMessage message, AckReplyConsumer consumer) -> { - EventMessage event = gson.fromJson(message.getData().toStringUtf8(), EventMessage.class); - messageProcessor.accept(event); - consumer.ack(); - }; - try { - subscriber = subscriberProvider.get(topic, receiver); + subscriber = subscriberProvider.get(topic, getMessageReceiver()); subscriber .startAsync() .awaitRunning(config.getSubscribtionTimeoutInSeconds(), TimeUnit.SECONDS); @@ -98,4 +95,21 @@ logger.atSevere().withCause(e).log("Timeout during subscriber shutdown"); } } + + @VisibleForTesting + MessageReceiver getMessageReceiver() { + return (PubsubMessage message, AckReplyConsumer consumer) -> { + try { + EventMessage event = gson.fromJson(message.getData().toStringUtf8(), EventMessage.class); + messageProcessor.accept(event); + consumer.ack(); + subscriberMetrics.incrementSucceedToConsumeMessage(); + } catch (Exception e) { + logger.atSevere().withCause(e).log( + "Exception when consuming message %s from topic %s [message: %s]", + message.getMessageId(), topic, message.getData().toStringUtf8()); + subscriberMetrics.incrementFailedToConsumeMessage(); + } + }; + } }
diff --git a/src/main/java/com/googlesource/gerrit/plugins/pubsub/PubSubEventsMetrics.java b/src/main/java/com/googlesource/gerrit/plugins/pubsub/PubSubEventsMetrics.java new file mode 100644 index 0000000..44968f4 --- /dev/null +++ b/src/main/java/com/googlesource/gerrit/plugins/pubsub/PubSubEventsMetrics.java
@@ -0,0 +1,35 @@ +// Copyright (C) 2021 The Android Open Source Project +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package com.googlesource.gerrit.plugins.pubsub; + +import com.google.gerrit.metrics.Description; +import com.google.gerrit.metrics.Field; +import com.google.gerrit.server.logging.PluginMetadata; + +public abstract class PubSubEventsMetrics { + + public Field<String> stringField(String metadataKey, String description) { + return Field.ofString( + metadataKey, + (metadataBuilder, fieldValue) -> + metadataBuilder.addPluginMetadata(PluginMetadata.create(metadataKey, fieldValue))) + .description(description) + .build(); + } + + public Description rateDescription(String unit, String description) { + return new Description(description).setRate().setUnit(unit); + } +}
diff --git a/src/main/java/com/googlesource/gerrit/plugins/pubsub/PubSubPublisher.java b/src/main/java/com/googlesource/gerrit/plugins/pubsub/PubSubPublisher.java index 24ef602..a802d71 100644 --- a/src/main/java/com/googlesource/gerrit/plugins/pubsub/PubSubPublisher.java +++ b/src/main/java/com/googlesource/gerrit/plugins/pubsub/PubSubPublisher.java
@@ -16,8 +16,11 @@ import com.gerritforge.gerrit.eventbroker.EventMessage; import com.google.api.core.ApiFuture; +import com.google.api.core.ApiFutureCallback; +import com.google.api.core.ApiFutures; import com.google.cloud.pubsub.v1.Publisher; import com.google.common.flogger.FluentLogger; +import com.google.common.util.concurrent.MoreExecutors; import com.google.gerrit.server.events.Event; import com.google.gson.Gson; import com.google.inject.Inject; @@ -37,6 +40,8 @@ } private final Gson gson; + private final PubSubPublisherMetrics publisherMetrics; + private final String topic; private final Publisher publisher; private final PubSubConfiguration pubSubProperties; @@ -45,9 +50,12 @@ PubSubConfiguration pubSubProperties, PublisherProvider publisherProvider, Gson gson, + PubSubPublisherMetrics publisherMetrics, @Assisted String topic) throws IOException { this.gson = gson; + this.publisherMetrics = publisherMetrics; + this.topic = topic; this.publisher = publisherProvider.get(topic); this.pubSubProperties = pubSubProperties; } @@ -72,7 +80,29 @@ } private ApiFuture<String> publishAsync(PubsubMessage pubsubMessage) { - return publisher.publish(pubsubMessage); + ApiFuture<String> publish = publisher.publish(pubsubMessage); + ApiFutures.addCallback( + publish, + new ApiFutureCallback<String>() { + @Override + public void onFailure(Throwable t) { + logger.atSevere().withCause(t).log( + "Exception when publishing message (id:%s) to topic '%s' [message: %s]", + pubsubMessage.getMessageId(), topic, pubsubMessage.getData().toStringUtf8()); + publisherMetrics.incrementFailedToPublishMessage(); + } + + @Override + public void onSuccess(String messageId) { + logger.atFine().log( + "Successfully published message (id:%s) to topic '%s' [message: %s]", + messageId, topic, pubsubMessage.getData().toStringUtf8()); + + publisherMetrics.incrementSucceedToPublishMessage(); + } + }, + MoreExecutors.directExecutor()); + return publish; } private void publishSync(PubsubMessage pubsubMessage) {
diff --git a/src/main/java/com/googlesource/gerrit/plugins/pubsub/PubSubPublisherMetrics.java b/src/main/java/com/googlesource/gerrit/plugins/pubsub/PubSubPublisherMetrics.java new file mode 100644 index 0000000..a205238 --- /dev/null +++ b/src/main/java/com/googlesource/gerrit/plugins/pubsub/PubSubPublisherMetrics.java
@@ -0,0 +1,52 @@ +// Copyright (C) 2021 The Android Open Source Project +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. +package com.googlesource.gerrit.plugins.pubsub; + +import com.google.gerrit.extensions.annotations.PluginName; +import com.google.gerrit.metrics.Counter1; +import com.google.gerrit.metrics.MetricMaker; +import com.google.inject.Inject; +import com.google.inject.Singleton; + +@Singleton +public class PubSubPublisherMetrics extends PubSubEventsMetrics { + private static final String PUBLISHER_SUCCESS_COUNTER = "publisher_success_counter"; + private static final String PUBLISHER_FAILURE_COUNTER = "publisher_failure_counter"; + + private final Counter1<String> publisherSuccessCounter; + private final Counter1<String> publisherFailureCounter; + + @Inject + public PubSubPublisherMetrics(MetricMaker metricMaker, @PluginName String pluginName) { + + this.publisherSuccessCounter = + metricMaker.newCounter( + String.join("/", pluginName, PUBLISHER_SUCCESS_COUNTER), + rateDescription("messages", "Number of successfully published messages"), + stringField(PUBLISHER_SUCCESS_COUNTER, "Count of published messages")); + this.publisherFailureCounter = + metricMaker.newCounter( + String.join("/", pluginName, PUBLISHER_FAILURE_COUNTER), + rateDescription("errors", "Number of messages failed to publish"), + stringField(PUBLISHER_FAILURE_COUNTER, "Count of messages failed to publish")); + } + + public void incrementSucceedToPublishMessage() { + publisherSuccessCounter.increment(PUBLISHER_SUCCESS_COUNTER); + } + + public void incrementFailedToPublishMessage() { + publisherFailureCounter.increment(PUBLISHER_FAILURE_COUNTER); + } +}
diff --git a/src/main/java/com/googlesource/gerrit/plugins/pubsub/PubSubSubscriberMetrics.java b/src/main/java/com/googlesource/gerrit/plugins/pubsub/PubSubSubscriberMetrics.java new file mode 100644 index 0000000..bbde683 --- /dev/null +++ b/src/main/java/com/googlesource/gerrit/plugins/pubsub/PubSubSubscriberMetrics.java
@@ -0,0 +1,53 @@ +// Copyright (C) 2021 The Android Open Source Project +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. +package com.googlesource.gerrit.plugins.pubsub; + +import com.google.gerrit.extensions.annotations.PluginName; +import com.google.gerrit.metrics.Counter1; +import com.google.gerrit.metrics.MetricMaker; +import com.google.inject.Inject; +import com.google.inject.Singleton; + +@Singleton +class PubSubSubscriberMetrics extends PubSubEventsMetrics { + + private static final String SUBSCRIBER_SUCCESS_COUNTER = "subscriber_success_counter"; + private static final String SUBSCRIBER_FAILURE_COUNTER = "subscriber_failure_counter"; + + private final Counter1<String> subscriberSuccessCounter; + private final Counter1<String> subscriberFailureCounter; + + @Inject + public PubSubSubscriberMetrics(MetricMaker metricMaker, @PluginName String pluginName) { + this.subscriberSuccessCounter = + metricMaker.newCounter( + String.join("/", pluginName, SUBSCRIBER_SUCCESS_COUNTER), + rateDescription( + "messages", "Number of messages successfully consumed by the subscriber"), + stringField(SUBSCRIBER_SUCCESS_COUNTER, "Count of successfully consumed messages")); + this.subscriberFailureCounter = + metricMaker.newCounter( + String.join("/", pluginName, SUBSCRIBER_FAILURE_COUNTER), + rateDescription("errors", "Number of messages failed to consume by the subscriber"), + stringField(SUBSCRIBER_FAILURE_COUNTER, "Count of messages failed to consume")); + } + + public void incrementFailedToConsumeMessage() { + subscriberFailureCounter.increment(SUBSCRIBER_FAILURE_COUNTER); + } + + public void incrementSucceedToConsumeMessage() { + subscriberSuccessCounter.increment(SUBSCRIBER_SUCCESS_COUNTER); + } +}
diff --git a/src/test/java/com/googlesource/gerrit/plugins/pubsub/PubSubEventSubscriberTest.java b/src/test/java/com/googlesource/gerrit/plugins/pubsub/PubSubEventSubscriberTest.java new file mode 100644 index 0000000..ec2dd69 --- /dev/null +++ b/src/test/java/com/googlesource/gerrit/plugins/pubsub/PubSubEventSubscriberTest.java
@@ -0,0 +1,74 @@ +// Copyright (C) 2021 The Android Open Source Project +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package com.googlesource.gerrit.plugins.pubsub; + +import static org.mockito.Mockito.only; +import static org.mockito.Mockito.verify; + +import com.gerritforge.gerrit.eventbroker.EventMessage; +import com.google.cloud.pubsub.v1.AckReplyConsumer; +import com.google.cloud.pubsub.v1.MessageReceiver; +import com.google.gerrit.json.OutputFormat; +import com.google.pubsub.v1.PubsubMessage; +import java.util.function.Consumer; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.mockito.Mock; +import org.mockito.junit.MockitoJUnitRunner; + +@RunWith(MockitoJUnitRunner.class) +public class PubSubEventSubscriberTest { + + @Mock PubSubConfiguration confMock; + @Mock SubscriberProvider subscriberProviderMock; + @Mock PubSubSubscriberMetrics pubSubSubscriberMetricsMock; + @Mock AckReplyConsumer ackReplyConsumerMock; + + private static final String TOPIC = "foo"; + + @Test + public void shouldIncrementFailedToConsumeMessageWhenReceivingFails() { + Consumer<EventMessage> failingConsumer = + (message) -> { + throw new RuntimeException("Error receiving message"); + }; + + messageReceiver(failingConsumer) + .receiveMessage(PubsubMessage.getDefaultInstance(), ackReplyConsumerMock); + + verify(pubSubSubscriberMetricsMock, only()).incrementFailedToConsumeMessage(); + } + + @Test + public void shouldIncrementSucceedToConsumeMessageWhenReceivingSucceeds() { + Consumer<EventMessage> succeedingConsumer = (message) -> {}; + + messageReceiver(succeedingConsumer) + .receiveMessage(PubsubMessage.getDefaultInstance(), ackReplyConsumerMock); + + verify(pubSubSubscriberMetricsMock, only()).incrementSucceedToConsumeMessage(); + } + + private MessageReceiver messageReceiver(Consumer<EventMessage> consumer) { + return new PubSubEventSubscriber( + OutputFormat.JSON_COMPACT.newGson(), + subscriberProviderMock, + confMock, + pubSubSubscriberMetricsMock, + TOPIC, + consumer) + .getMessageReceiver(); + } +}
diff --git a/src/test/java/com/googlesource/gerrit/plugins/pubsub/PubSubPublisherTest.java b/src/test/java/com/googlesource/gerrit/plugins/pubsub/PubSubPublisherTest.java new file mode 100644 index 0000000..c009973 --- /dev/null +++ b/src/test/java/com/googlesource/gerrit/plugins/pubsub/PubSubPublisherTest.java
@@ -0,0 +1,102 @@ +// Copyright (C) 2021 The Android Open Source Project +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package com.googlesource.gerrit.plugins.pubsub; + +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.only; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +import com.gerritforge.gerrit.eventbroker.EventMessage; +import com.google.api.core.ApiFutures; +import com.google.cloud.pubsub.v1.Publisher; +import com.google.gerrit.json.OutputFormat; +import com.google.gerrit.server.events.ProjectCreatedEvent; +import java.io.IOException; +import java.util.UUID; +import org.junit.Before; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.mockito.Mock; +import org.mockito.junit.MockitoJUnitRunner; + +@RunWith(MockitoJUnitRunner.class) +public class PubSubPublisherTest { + PubSubPublisher objectUnderTest; + + @Mock PubSubConfiguration confMock; + @Mock PublisherProvider publisherProviderMock; + @Mock Publisher publisherMock; + @Mock PubSubPublisherMetrics pubSubPublisherMetricsMock; + + private static final String TOPIC = "foo"; + private static final EventMessage eventMessage = + new EventMessage( + new EventMessage.Header(UUID.randomUUID(), UUID.randomUUID()), new ProjectCreatedEvent()); + + @Before + public void setUp() throws IOException { + when(publisherProviderMock.get(TOPIC)).thenReturn(publisherMock); + objectUnderTest = + new PubSubPublisher( + confMock, + publisherProviderMock, + OutputFormat.JSON_COMPACT.newGson(), + pubSubPublisherMetricsMock, + TOPIC); + } + + @Test + public void shouldIncrementFailedToPublishMessageWhenAsyncPublishFails() { + when(confMock.isSendAsync()).thenReturn(true); + when(publisherMock.publish(any())) + .thenReturn(ApiFutures.immediateFailedFuture(new Exception("Something went wrong"))); + + objectUnderTest.publish(eventMessage); + + verify(pubSubPublisherMetricsMock, only()).incrementFailedToPublishMessage(); + } + + @Test + public void shouldIncrementFailedToPublishMessageWhenSyncPublishFails() { + when(confMock.isSendAsync()).thenReturn(false); + when(publisherMock.publish(any())) + .thenReturn(ApiFutures.immediateFailedFuture(new Exception("Something went wrong"))); + + objectUnderTest.publish(eventMessage); + + verify(pubSubPublisherMetricsMock, only()).incrementFailedToPublishMessage(); + } + + @Test + public void shouldIncrementSuccessToPublishMessageWhenAsyncPublishSucceeds() { + when(confMock.isSendAsync()).thenReturn(true); + when(publisherMock.publish(any())).thenReturn(ApiFutures.immediateFuture("some-message-id")); + + objectUnderTest.publish(eventMessage); + + verify(pubSubPublisherMetricsMock, only()).incrementSucceedToPublishMessage(); + } + + @Test + public void shouldIncrementSuccessToPublishMessageWhenSyncPublishSucceeds() { + when(confMock.isSendAsync()).thenReturn(false); + when(publisherMock.publish(any())).thenReturn(ApiFutures.immediateFuture("some-message-id")); + + objectUnderTest.publish(eventMessage); + + verify(pubSubPublisherMetricsMock, only()).incrementSucceedToPublishMessage(); + } +}