Merge pull request #150 from RADAR-base/feature/sentry

Add support for Sentry monitoring

Author: pvannierop

README.md

@@ -1,7 +1,19 @@
 # Kafka Connect REST Source and Fitbit Source
 
-This project contains a Kafka Connect source connector for a general REST API, and one for
-Fitbit in particular. The documentation of the Kafka Connect REST source still needs to be done.
+This project contains a Kafka Connect source connector for a general REST API, for
+specific Fitbit and Oura devices. The documentation of the Kafka Connect REST source still needs to
+be done.
+
+<!-- TOC -->
+
+* [Kafka Connect REST Source and Fitbit Source](#kafka-connect-rest-source-and-fitbit-source)
+  * [Fitbit source connector](#fitbit-source-connector)
+    * [Installation](#installation)
+    * [Usage](#usage)
+  * [Sentry monitoring](#sentry-monitoring)
+  * [Contributing](#contributing)
+
+<!-- TOC -->
 
 ## Fitbit source connector
 
@@ -12,7 +24,9 @@ of Java 17 or later.
 
 ### Usage
 
-Generally, this component is installed with [RADAR-Kubernetes](https://github.com/RADAR-base/RADAR-Kubernetes). It uses Docker image [radarbase/kafka-connect-rest-fitbit-source](https://hub.docker.com/r/radarbase/kafka-connect-rest-fitbit-source).
+Generally, this component is installed
+with [RADAR-Kubernetes](https://github.com/RADAR-base/RADAR-Kubernetes). It uses Docker
+image [radarbase/kafka-connect-rest-fitbit-source](https://hub.docker.com/r/radarbase/kafka-connect-rest-fitbit-source).
 
 First, [register a Fitbit App](https://dev.fitbit.com/apps) with Fitbit. It should be either a
 server app, for multiple users, or a personal app for a single user. With the server app, you need
@@ -22,7 +36,8 @@ For every Fitbit user you want access to, copy `docker/fitbit-user.yml.template`
 `docker/users/`. Get an access token and refresh token for the user using for example the
 [Fitbit OAuth 2.0 tutorial page](https://dev.fitbit.com/apps/oauthinteractivetutorial).
 
-For automatic configuration for multiple users, please take a look at `scripts/REDCAP-FITBIT-AUTH-AUTO/README.md`.
+For automatic configuration for multiple users, please take a look at
+`scripts/REDCAP-FITBIT-AUTH-AUTO/README.md`.
 
 Copy `docker/source-fitbit.properties.template` to `docker/source-fitbit.properties` and enter
 your Fitbit App client ID and client secret. The following tables shows the possible properties.
@@ -37,6 +52,10 @@ your Fitbit App client ID and client secret. The following tables shows the poss
 <th>Importance</th>
 </tr>
 <tr>
+<td>application.loop.interval.ms</td><td>How often to perform the main application loop (only controls how often to poll for new user registrations).></td><td>long</td><td>300000</td><td></td><td></td></tr>
+<tr>
+<td>user.cache.refresh.interval.ms</td><td>How often to invalidate the cache and poll for new user registrations.</td><td>long</td><td>3600000</td><td></td><td></td></tr>
+<tr>
 <td>rest.source.poll.interval.ms</td><td>How often to poll the source URL.</td><td>long</td><td>60000</td><td></td><td>low</td></tr>
 <tr>
 <td>rest.source.base.url</td><td>Base URL for REST source connector.</td><td>string</td><td></td><td></td><td>high</td></tr>
@@ -90,7 +109,8 @@ your Fitbit App client ID and client secret. The following tables shows the poss
 <td>fitbit.user.firebase.collection.user.name</td><td>Firestore Collection for retrieving User details. Only used when a Firebase based user repository is used.</td><td>string</td><td>users</td><td></td><td>low</td></tr>
 </tbody></table>
 
-If the ManagementPortal is used to authenticate against the user repository, please add an OAuth client to ManagementPortal with the following properties:
+If the ManagementPortal is used to authenticate against the user repository, please add an OAuth
+client to ManagementPortal with the following properties:
 
 ```
 Client ID: fitbit.user.repository.client.id
@@ -102,7 +122,8 @@ Access Token validity: 600
 Refresh Token validity: 0
 ```
 
-Finally set the `fitbit.user.repository.oauth.token.url` to `http://managementportal-app:8080/managementportal/oauth/token`.
+Finally set the `fitbit.user.repository.oauth.token.url` to
+`http://managementportal-app:8080/managementportal/oauth/token`.
 
 Now you can run a full Kafka stack using
 
@@ -159,7 +180,29 @@ sequenceDiagram
   connector ->> connector: Update offset times
 ```
 
+## Sentry monitoring
+
+To enable Sentry monitoring for the generic REST, Fitbit, or Oura source connector service:
+
+1. Set a `SENTRY_DSN` environment variable that points to the desired Sentry DSN.
+2. (Optional) Set the `SENTRY_LOG_LEVEL` environment variable to control the minimum log level of
+   events sent to Sentry.
+   The default log level for Sentry is `WARN`. Possible values are `TRACE`, `DEBUG`, `INFO`, `WARN`,
+   and `ERROR`.
+
+For further configuration of Sentry via environmental variables see [here](https://docs.sentry.io/platforms/java/configuration/#configuration-via-the-runtime-environment). For instance:
+
+```
+SENTRY_LOG_LEVEL: 'ERROR'
+SENTRY_DSN: 'https://000000000000.ingest.de.sentry.io/000000000000'
+SENTRY_ATTACHSTACKTRACE: true
+SENTRY_STACKTRACE_APP_PACKAGES: io.confluent.connect,org.radarbase.connect.rest
+```
+
 ## Contributing
 
-Code should be formatted using the [Google Java Code Style Guide](https://google.github.io/styleguide/javaguide.html).
-If you want to contribute a feature or fix browse our [issues](https://github.com/RADAR-base/RADAR-REST-Connector/issues), and please make a pull request.
+Code should be formatted using
+the [Google Java Code Style Guide](https://google.github.io/styleguide/javaguide.html).
+If you want to contribute a feature or fix browse
+our [issues](https://github.com/RADAR-base/RADAR-REST-Connector/issues), and please make a pull
+request.

buildSrc/src/main/kotlin/Versions.kt

@@ -6,7 +6,7 @@ object Versions {
     const val kotlin = "1.9.22"
     const val wrapper = "8.4"
 
-    const val radarCommons = "1.1.2"
+    const val radarCommons = "1.1.3"
     const val confluent = "7.7.0"
     const val kafka = "$confluent-ce"
     const val avro = "1.12.0"
@@ -18,6 +18,7 @@ object Versions {
 
     const val log4j2 = "2.23.1"
     const val slf4j = "2.0.13"
+    const val sentryLog4j = "1.7.30"
 
     const val okhttp = "4.12.0"
 

docker-compose.yml

@@ -180,6 +180,10 @@ services:
       KAFKA_HEAP_OPTS: "-Xms256m -Xmx768m"
       KAFKA_BROKERS: 3
       CONNECT_LOG4J_LOGGERS: "org.reflections=ERROR"
+      # SENTRY_LOG_LEVEL: 'ERROR'
+      # SENTRY_DSN: 'https://000000000000.ingest.de.sentry.io/000000000000'
+      # SENTRY_ATTACHSTACKTRACE: true
+      # SENTRY_STACKTRACE_APP_PACKAGES: io.confluent.connect,org.radarbase.connect.rest
 
   #---------------------------------------------------------------------------#
   # RADAR Oura connector                                                     #
@@ -221,3 +225,7 @@ services:
       KAFKA_HEAP_OPTS: "-Xms256m -Xmx768m"
       KAFKA_BROKERS: 3
       CONNECT_LOG4J_LOGGERS: "org.reflections=ERROR"
+      # SENTRY_LOG_LEVEL: 'ERROR'
+      # SENTRY_DSN: 'https://000000000000.ingest.de.sentry.io/000000000000'
+      # SENTRY_ATTACHSTACKTRACE: true
+      # SENTRY_STACKTRACE_APP_PACKAGES: io.confluent.connect,org.radarbase.connect.rest

docker/kafka-wait docker/ensuredocker/kafka-wait renamed to docker/ensure

@@ -7,9 +7,10 @@ fi
 
 max_timeout=32
 
-
 IS_TEMP=0
 
+echo "===> Wait for infrastructure ..."
+
 if [ -z "$COMMAND_CONFIG_FILE_PATH" ]; then
     COMMAND_CONFIG_FILE_PATH="$(mktemp)"
     IS_TEMP=1

docker/launch

@@ -33,14 +33,6 @@ if [ "$KAFKA_JMX_PORT" ]; then
   export JMX_PORT=$KAFKA_JMX_PORT
   export KAFKA_JMX_OPTS="$KAFKA_JMX_OPTS -Djava.rmi.server.hostname=$KAFKA_JMX_HOSTNAME -Dcom.sun.management.jmxremote.local.only=false -Dcom.sun.management.jmxremote.rmi.port=$JMX_PORT -Dcom.sun.management.jmxremote.port=$JMX_PORT"
 fi
-#
-## Busy waiting loop that waits until all topic are available
-echo "===> Wait for infrastructure ..."
-kafka-wait
-radar_check=$?
-if [ "$radar_check" -ne 0 ]; then
-    exit $radar_check
-fi
 
 echo "===> Launching ${COMPONENT} ..."
 # Add our jar to the classpath so that the custom classes can be loaded first.

docker/log4j.properties.template

@@ -0,0 +1,32 @@
+# This template file was taken from the Confluent Platform distribution and modified to add Sentry support in Docker images.
+# See: https://docs.confluent.io/platform/current/installation/docker/development.html#log-to-external-volumes
+
+log4j.rootLogger={{ env["CONNECT_LOG4J_ROOT_LOGLEVEL"] | default('INFO') }}, stdout{% if env['SENTRY_DSN'] %}, sentryAppender{% endif %}
+
+
+log4j.appender.stdout=org.apache.log4j.ConsoleAppender
+log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
+log4j.appender.stdout.layout.ConversionPattern ={{ env["CONNECT_LOG4J_APPENDER_STDOUT_LAYOUT_CONVERSIONPATTERN"] | default('[%d] %p %m (%c)%n') }}
+
+# Appender for Sentry monitoring
+{% if env['SENTRY_DSN'] %}
+log4j.appender.sentryAppender=io.sentry.log4j.SentryAppender
+log4j.appender.sentryAppender.threshold={{ env['SENTRY_LOG_LEVEL'] | default('ERROR') }}
+{% endif %}
+
+{% set default_loggers = {
+'org.reflections': 'ERROR',
+'org.apache.zookeeper': 'ERROR',
+'org.I0Itec.zkclient': 'ERROR'
+} -%}
+
+{% if env['CONNECT_LOG4J_LOGGERS'] %}
+# loggers from CONNECT_LOG4J_LOGGERS env variable
+{% set loggers = parse_log4j_loggers(env['CONNECT_LOG4J_LOGGERS']) %}
+{% else %}
+# default log levels
+{% set loggers = default_loggers %}
+{% endif %}
+{% for logger,loglevel in loggers.items() %}
+log4j.logger.{{logger}}={{loglevel}}
+{% endfor %}

kafka-connect-fitbit-source/Dockerfile

@@ -32,7 +32,7 @@ COPY ./kafka-connect-fitbit-source/src/ /code/kafka-connect-fitbit-source/src
 
 RUN gradle jar
 
-FROM confluentinc/cp-kafka-connect-base:7.5.0
+FROM confluentinc/cp-kafka-connect-base:7.6.0
 
 USER root
 
@@ -41,7 +41,7 @@ RUN yum install -y zulu17-ca-jdk-headless && yum install -y zulu17-ca-jre-headle
 
 USER appuser
 
-MAINTAINER Joris Borgdorff <joris@thehyve.nl>
+MAINTAINER Pim van Nierop <pim@thehyve.nl>
 
 LABEL description="Kafka REST API Source connector"
 
@@ -57,7 +57,13 @@ COPY --from=builder /code/kafka-connect-rest-source/build/libs/*.jar ${CONNECT_P
 COPY --from=builder /code/kafka-connect-fitbit-source/build/libs/*.jar ${CONNECT_PLUGIN_PATH}/kafka-connect-fitbit-source/
 
 # Load topics validator
-COPY  --chown=appuser:appuser ./docker/kafka-wait /usr/bin/kafka-wait
+COPY  --chown=appuser:appuser ./docker/ensure /etc/confluent/docker/ensure
 
 # Load modified launcher
 COPY  --chown=appuser:appuser ./docker/launch /etc/confluent/docker/launch
+
+# Overwrite the log4j configuration to include Sentry monitoring.
+COPY ./docker/log4j.properties.template /etc/confluent/docker/log4j.properties.template
+# Copy Sentry monitoring jars.
+COPY --from=builder /code/kafka-connect-fitbit-source/build/third-party/sentry-* /etc/kafka-connect/jars
+

kafka-connect-fitbit-source/build.gradle.kts

@@ -27,5 +27,12 @@ dependencies {
     compileOnly(platform("com.fasterxml.jackson:jackson-bom:${Versions.jackson}"))
     compileOnly("com.fasterxml.jackson.core:jackson-databind")
 
+    // Application monitoring
+    // This dependency is not used by the REST connector, but copied into the Docker image (Dockerfile)
+    compileOnly("io.sentry:sentry-log4j:${Versions.sentryLog4j}") {
+        // Exclude log4j with security vulnerability (safe version is provided by docker image).
+        exclude(group = "log4j", module = "log4j")
+    }
+
     testImplementation("org.apache.kafka:connect-api:${Versions.kafka}")
 }

kafka-connect-fitbit-source/src/main/java/org/radarbase/connect/rest/fitbit/FitbitSourceConnector.java

@@ -20,6 +20,7 @@
 import static org.radarbase.connect.rest.fitbit.FitbitRestSourceConnectorConfig.FITBIT_USERS_CONFIG;
 
 import java.io.IOException;
+import java.time.Duration;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
@@ -45,9 +46,12 @@ public class FitbitSourceConnector extends AbstractRestSourceConnector {
 
   @Override
   public void start(Map<String, String> props) {
+    logger.info("Starting Fitbit source connector");
     super.start(props);
     executor = Executors.newSingleThreadScheduledExecutor();
 
+    Duration applicationLoopInterval = config.getApplicationLoopInterval();
+
     executor.scheduleAtFixedRate(() -> {
       if (repository.hasPendingUpdates()) {
         try {
@@ -66,7 +70,7 @@ public void start(Map<String, String> props) {
       } else {
         logger.info("No pending updates found. Not attempting to refresh users.");
       }
-    }, 0, 5, TimeUnit.MINUTES);
+    }, 0, applicationLoopInterval.toSeconds(), TimeUnit.SECONDS);
   }
 
   @Override

kafka-connect-fitbit-source/src/main/java/org/radarbase/connect/rest/fitbit/route/FitbitPollingRoute.java

@@ -100,7 +100,7 @@ public abstract class FitbitPollingRoute implements PollingRequestRoute {
   protected static final TemporalAmount ONE_SECOND = SECONDS.getDuration();
   protected static final TemporalAmount ONE_MINUTE = MINUTES.getDuration();
 
-  private static final Logger logger = LoggerFactory.getLogger(FitbitSleepRoute.class);
+  private static final Logger logger = LoggerFactory.getLogger(FitbitPollingRoute.class);
 
   /** Committed offsets. */
   private Map<String, Instant> offsets;
@@ -164,7 +164,7 @@ public void requestEmpty(RestRequest request) {
   @Override
   public void requestFailed(RestRequest request, Response response) {
     if (response != null && response.code() == 429) {
-      User user = ((FitbitRestRequest)request).getUser();
+      User user = ((FitbitRestRequest) request).getUser();
       tooManyRequestsForUser.add(user);
       String cooldownString = response.header("Retry-After");
       Duration cooldown = getTooManyRequestsCooldown();
@@ -179,6 +179,8 @@ public void requestFailed(RestRequest request, Response response) {
       lastPollPerUser.put(user.getId(), backOff);
       logger.info("Too many requests for user {}. Backing off until {}",
           user, backOff.plus(getPollIntervalPerUser()));
+    } else if (response != null) {
+      logger.warn("Failed to make request {}. Response is: {}", request, response);
     } else {
       logger.warn("Failed to make request {}", request);
     }
@@ -197,8 +199,11 @@ public Stream<FitbitRestRequest> requests() {
     lastPoll = Instant.now();
     try {
       return userRepository.stream()
+          // Collect Instant of nextPoll for each user
           .map(u -> new AbstractMap.SimpleImmutableEntry<>(u, nextPoll(u)))
+          // Keep users where the lastPoll is later than the nextPoll for the user (i.e., user needs to be polled)
           .filter(u -> lastPoll.isAfter(u.getValue()))
+          // Sort users by nextPoll (old to new?)
           .sorted(Map.Entry.comparingByValue())
           .flatMap(u -> this.createRequests(u.getKey()))
           .filter(Objects::nonNull);