Merge pull request #163 from RADAR-base/release-0.6.2

Release 0.6.2

Author: mpgxvii

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.
@@ -94,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
@@ -106,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
 
@@ -163,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

@@ -1,6 +1,6 @@
 @Suppress("ConstPropertyName", "MemberVisibilityCanBePrivate")
 object Versions {
-    const val project = "0.6.1"
+    const val project = "0.6.2"
 
     const val java = 17
     const val kotlin = "1.9.22"
@@ -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

@@ -52,7 +52,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

@@ -34,5 +34,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/user/ServiceUserRepository.kt

@@ -90,6 +90,8 @@ class ServiceUserRepository : UserRepository {
             tokenUrl = URLBuilder(config.fitbitUserRepositoryTokenUrl.toString()).build(),
             clientId = config.fitbitUserRepositoryClientId,
             clientSecret = config.fitbitUserRepositoryClientSecret,
+            scope = "SUBJECT.READ MEASUREMENT.CREATE",
+            audience = "res_restAuthorizer",
         )
 
         val refreshDuration = config.userCacheRefreshInterval.toKotlinDuration()
@@ -113,6 +115,8 @@ class ServiceUserRepository : UserRepository {
         tokenUrl: Url?,
         clientId: String?,
         clientSecret: String?,
+        scope: String?,
+        audience: String?,
     ): HttpClient = HttpClient(CIO) {
         if (tokenUrl != null) {
             install(Auth) {
@@ -121,6 +125,8 @@ class ServiceUserRepository : UserRepository {
                         tokenUrl.toString(),
                         clientId,
                         clientSecret,
+                        scope,
+                        audience,
                     ).copyWithEnv("MANAGEMENT_PORTAL"),
                     baseUrl.host,
                 )

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

@@ -36,7 +36,8 @@
  import java.util.stream.Collectors;
  import java.util.stream.Stream;
  import okhttp3.Credentials;
- import okhttp3.HttpUrl;
+import okhttp3.FormBody;
+import okhttp3.HttpUrl;
  import okhttp3.MediaType;
  import okhttp3.OkHttpClient;
  import okhttp3.Request;
@@ -64,6 +65,9 @@ public class ServiceUserRepositoryLegacy implements UserRepository {
    private static final Duration CONNECTION_TIMEOUT = Duration.ofSeconds(60);
    private static final Duration CONNECTION_READ_TIMEOUT = Duration.ofSeconds(90);
 
+   private static final String CLIENT_AUDIENCE = "res_restAuthorizer"; 
+   private static final String CLIENT_AUDIENCE_KEY = "audience";
+
    private final OkHttpClient client;
    private final Map<String, OAuth2UserCredentials> cachedCredentials;
    private final AtomicReference<Instant> nextFetch = new AtomicReference<>(MIN_INSTANT);
@@ -76,6 +80,19 @@ public class ServiceUserRepositoryLegacy implements UserRepository {
 
    public ServiceUserRepositoryLegacy() {
      this.client = new OkHttpClient.Builder()
+          .addInterceptor(chain -> {
+            Request req = chain.request();
+            if ("POST".equalsIgnoreCase(req.method()) && req.body() instanceof FormBody) {
+                FormBody oldBody = (FormBody) req.body();
+                FormBody.Builder newBody = new FormBody.Builder();
+                for (int i = 0; i < oldBody.size(); i++) {
+                    newBody.addEncoded(oldBody.encodedName(i), oldBody.encodedValue(i));
+                }
+                newBody.add(CLIENT_AUDIENCE_KEY, CLIENT_AUDIENCE);
+                req = req.newBuilder().post(newBody.build()).build();
+            }
+            return chain.proceed(req);
+          })
          .connectTimeout(CONNECTION_TIMEOUT)
          .readTimeout(CONNECTION_READ_TIMEOUT)
          .build();