Merge pull request #37 from RADAR-base/multipe_user_repos

Multipe user repos

Author: yatharthranjan

.gitignore

@@ -6,3 +6,4 @@ out/
 .gradle/
 docker/users
 docker/source-fitbit.properties
+docker/rest-source-authorizer/etc/rest_source_clients_configs.yml

README.md

@@ -55,7 +55,7 @@ your Fitbit App client ID and client secret. The following tables shows the poss
 <tr>
 <td>fitbit.api.intraday</td></td><td>Set to true if the client has permissions to Fitbit Intraday API, false otherwise.</td></td><td>boolean</td></td><td>true</td></td><td></td></td><td>medium</td></td></tr>
 <tr>
-<td>fitbit.user.repository.class</td></td><td>Class for managing users and authentication.</td></td><td>class</td></td><td>org.radarbase.connect.rest.fitbit.user.YamlUserRepository</td></td><td>Class extending org.radarbase.connect.rest.fitbit.user.UserRepository</td></td><td>medium</td></td></tr>
+<td>fitbit.user.repository.classes</td></td><td>Class(es) for managing users and authentication. Comma separate for using multiple repositories</td></td><td>list</td></td><td>org.radarbase.connect.rest.fitbit.user.YamlUserRepository</td></td><td>Classes extending org.radarbase.connect.rest.fitbit.user.UserRepository</td></td><td>medium</td></td></tr>
 <tr>
 <td>fitbit.user.dir</td></td><td>Directory containing Fitbit user information and credentials. Only used if a file-based user repository is configured.</td></td><td>string</td></td><td>/var/lib/kafka-connect-fitbit-source/users</td></td><td></td></td><td>low</td></td></tr>
 <tr>

docker-compose.yml

@@ -163,6 +163,8 @@ services:
       - kafka-2
       - kafka-3
       - schema-registry-1
+    networks:
+      - default
     environment:
       CONNECT_BOOTSTRAP_SERVERS: PLAINTEXT://kafka-1:9092,PLAINTEXT://kafka-2:9092,PLAINTEXT://kafka-3:9092
       CONNECT_REST_PORT: 8083
@@ -181,5 +183,6 @@ services:
       CONNECT_ZOOKEEPER_CONNECT: zookeeper-1:2181,zookeeper-2:2181,zookeeper-3:2181
       CONNECTOR_PROPERTY_FILE_PREFIX: "source-fitbit"
       KAFKA_HEAP_OPTS: "-Xms256m -Xmx768m"
+      CONNECT_LOG4J_ROOT_LOGLEVEL: DEBUG
       KAFKA_BROKERS: 3
       CONNECT_LOG4J_LOGGERS: "org.reflections=ERROR"

docker/rest-source-authorizer/docker-compose.yml

@@ -0,0 +1,34 @@
+---
+version: '2.4'
+
+services:
+  radar-rest-source-auth-backend:
+    image: radarbase/radar-rest-source-auth-backend:1.0.0
+    depends_on:
+      - auth-postgresql
+    environment:
+      - SPRING_DATASOURCE_URL=jdbc:postgresql://auth-postgresql:5432/restsourceauth
+      - SPRING_DATASOURCE_USERNAME=radarcns
+      - SPRING_DATASOURCE_PASSWORD=radarcns
+      - REST_SOURCE_AUTHORIZER_SOURCE_CLIENTS_FILE_PATH=app-includes/rest_source_clients_configs.yml
+      - LOGGING_LEVEL_ORG_SPRINGFRAMEWORK_WEB=DEBUG
+      - APP_SLEEP=10 # gives time for the database to boot before the application
+    ports:
+      - "8080:8080"
+    volumes:
+      - ./etc:/app-includes
+
+  auth-postgresql:
+    image: postgres:10.4-alpine
+    environment:
+      - POSTGRES_USER=radarcns
+      - POSTGRES_PASSWORD=radarcns
+      - POSTGRES_DB=restsourceauth
+
+  radar-rest-source-authorizer:
+    image: radarbase/radar-rest-source-authorizer:1.0.0
+    depends_on:
+      - radar-rest-source-auth-backend
+      - auth-postgresql
+    ports:
+      - "3000:80"

docker/rest-source-authorizer/etc/rest_source_clients_configs.yml.template

@@ -0,0 +1,7 @@
+rest_source_clients:
+  - source_type: FitBit
+    authorization_endpoint: https://www.fitbit.com/oauth2/authorize
+    token_endpoint: https://api.fitbit.com/oauth2/token
+    client_id: FITBITIT
+    client_secret: FITBITSECRET
+    scope: activity heartrate sleep profile

docker/source-fitbit.properties.template

@@ -7,5 +7,5 @@ rest.source.request.generator.class=org.radarbase.connect.rest.fitbit.request.Fi
 fitbit.api.client=?
 fitbit.api.secret=?
 fitbit.max.users.per.poll=10
-fitbit.user.repository.class=org.radarbase.connect.rest.fitbit.user.ServiceUserRepository
-fitbit.user.repository.url=http://radar-device-auth-backend:8080/
+fitbit.user.repository.classes=org.radarbase.connect.rest.fitbit.user.YamlUserRepository,org.radarbase.connect.rest.fitbit.user.ServiceUserRepository
+fitbit.user.repository.url=http://localhost:8080/

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

@@ -23,11 +23,8 @@
 import java.nio.charset.StandardCharsets;
 import java.nio.file.Path;
 import java.nio.file.Paths;
-import java.time.Duration;
-import java.util.Base64;
-import java.util.Collections;
-import java.util.List;
-import java.util.Map;
+import java.util.*;
+
 import okhttp3.Headers;
 import okhttp3.HttpUrl;
 import org.apache.kafka.common.config.ConfigDef;
@@ -40,7 +37,6 @@
 import org.apache.kafka.common.config.ConfigException;
 import org.apache.kafka.connect.errors.ConnectException;
 import org.radarbase.connect.rest.RestSourceConnectorConfig;
-import org.radarbase.connect.rest.config.ValidClass;
 import org.radarbase.connect.rest.fitbit.user.UserRepository;
 import org.radarbase.connect.rest.fitbit.user.YamlUserRepository;
 
@@ -61,9 +57,9 @@ public class FitbitRestSourceConnectorConfig extends RestSourceConnectorConfig {
   private static final String FITBIT_API_SECRET_DOC = "Secret for the Fitbit API client set in fitbit.api.client.";
   private static final String FITBIT_API_SECRET_DISPLAY = "Fitbit API client secret";
 
-  public static final String FITBIT_USER_REPOSITORY_CONFIG = "fitbit.user.repository.class";
-  private static final String FITBIT_USER_REPOSITORY_DOC = "Class for managing users and authentication.";
-  private static final String FITBIT_USER_REPOSITORY_DISPLAY = "User repository class";
+  public static final String FITBIT_USER_REPOSITORIES_CONFIG = "fitbit.user.repository.classes";
+  private static final String FITBIT_USER_REPOSITORIES_DOC = "Classes for managing users and authentication. Can be a Comma-separated list for multiple repositories.";
+  private static final String FITBIT_USER_REPOSITORIES_DISPLAY = "User repository classes";
 
   public static final String FITBIT_API_INTRADAY_ACCESS_CONFIG = "fitbit.api.intraday";
   private static final String FITBIT_API_INTRADAY_ACCESS_DOC = "Set to true if the client has permissions to Fitbit Intraday API, false otherwise.";
@@ -115,19 +111,22 @@ public class FitbitRestSourceConnectorConfig extends RestSourceConnectorConfig {
   private static final String FITBIT_ACTIVITY_LOG_TOPIC_DEFAULT = "connect_fitbit_activity_log";
   private static final String FITBIT_ACTIVITY_LOG_TOPIC_DISPLAY = "Activity log topic";
 
-  private final UserRepository userRepository;
+  private final List<UserRepository> userRepositories;
   private final Headers clientCredentials;
 
   @SuppressWarnings("unchecked")
   public FitbitRestSourceConnectorConfig(ConfigDef config, Map<String, String> parsedConfig, boolean doLog) {
     super(config, parsedConfig, doLog);
+    userRepositories = new ArrayList<>();
 
     try {
-      userRepository = ((Class<? extends UserRepository>)
-          getClass(FITBIT_USER_REPOSITORY_CONFIG)).getDeclaredConstructor().newInstance();
+      for(String c: getList(FITBIT_USER_REPOSITORIES_CONFIG)) {
+        userRepositories.add(
+            ((Class<? extends UserRepository>) Class.forName(c)).getDeclaredConstructor().newInstance());
+      }
     } catch (IllegalAccessException | InstantiationException
-        | InvocationTargetException | NoSuchMethodException e) {
-      throw new ConnectException("Invalid class for: " + SOURCE_PAYLOAD_CONVERTER_CONFIG, e);
+        | InvocationTargetException | NoSuchMethodException | ClassNotFoundException e) {
+      throw new ConnectException("Invalid class(es) for: " + FITBIT_USER_REPOSITORIES_CONFIG, e);
     }
 
     String credentialString = getFitbitClient() + ":" + getFitbitClientSecret();
@@ -194,16 +193,15 @@ public String toString() {
             Width.SHORT,
             FITBIT_API_INTRADAY_ACCESS_DISPLAY)
 
-        .define(FITBIT_USER_REPOSITORY_CONFIG,
-            Type.CLASS,
-            YamlUserRepository.class,
-            ValidClass.isSubclassOf(UserRepository.class),
+        .define(FITBIT_USER_REPOSITORIES_CONFIG,
+            Type.LIST,
+            YamlUserRepository.class.getName(),
             Importance.MEDIUM,
-            FITBIT_USER_REPOSITORY_DOC,
+            FITBIT_USER_REPOSITORIES_DOC,
             group,
             ++orderInGroup,
             Width.SHORT,
-            FITBIT_USER_REPOSITORY_DISPLAY)
+            FITBIT_USER_REPOSITORIES_DISPLAY)
 
         .define(FITBIT_USER_CREDENTIALS_DIR_CONFIG,
             Type.STRING,
@@ -316,9 +314,9 @@ public String getFitbitClientSecret() {
     return getPassword(FITBIT_API_SECRET_CONFIG).value();
   }
 
-  public UserRepository getUserRepository() {
-    userRepository.initialize(this);
-    return userRepository;
+  public List<UserRepository> getUserRepositories() {
+    userRepositories.forEach(u -> u.initialize(this));
+    return userRepositories;
   }
 
   public String getFitbitIntradayStepsTopic() {

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

@@ -17,23 +17,21 @@
 
 package org.radarbase.connect.rest.fitbit;
 
+import static java.util.stream.Collectors.toSet;
 import static org.radarbase.connect.rest.fitbit.FitbitRestSourceConnectorConfig.FITBIT_USERS_CONFIG;
 
 import java.io.IOException;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
+import java.util.*;
 import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
 import java.util.stream.Collectors;
-import java.util.stream.Stream;
 
 import org.apache.kafka.common.config.ConfigDef;
 import org.apache.kafka.common.config.ConfigException;
 import org.radarbase.connect.rest.AbstractRestSourceConnector;
 import org.radarbase.connect.rest.fitbit.user.User;
+import org.radarbase.connect.rest.fitbit.user.UserRepository;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -50,10 +48,13 @@ public void start(Map<String, String> props) {
     executor = Executors.newSingleThreadScheduledExecutor();
 
     executor.scheduleAtFixedRate(() -> {
+      Set<? extends User> newUsers = new HashSet<>();
       try {
         logger.info("Requesting latest user details...");
-        Set<? extends User> newUsers = getConfig(props, false).getUserRepository().stream()
-            .collect(Collectors.toSet());
+        for(UserRepository u: getConfig(props, false).getUserRepositories()) {
+          Set users = u.stream().collect(toSet());
+          newUsers.addAll(users);
+        }
         if (configuredUsers != null && !newUsers.equals(configuredUsers)) {
           logger.info("User info mismatch found. Requesting reconfiguration...");
           reconfigure();
@@ -95,26 +96,32 @@ private List<Map<String, String>> configureTasks(int maxTasks) {
     Map<String, String> baseConfig = config.originalsStrings();
     FitbitRestSourceConnectorConfig fitbitConfig = getConfig(baseConfig);
     // Divide the users over tasks
-    try {
+    List<Map<String, String>> userTasks = new ArrayList<>();
+    this.configuredUsers = new HashSet<>();
 
-      List<Map<String, String>> userTasks = fitbitConfig.getUserRepository().stream()
-          .map(User::getId)
-          // group users based on their hashCode
-          // in principle this allows for more efficient reconfigurations for a fixed number of tasks,
-          // since that allows existing tasks to only handle small modifications users to handle.
-          .collect(Collectors.groupingBy(
-              u -> Math.abs(u.hashCode()) % maxTasks,
-              Collectors.joining(",")))
-          .values().stream()
-          .map(u -> {
-            Map<String, String> config = new HashMap<>(baseConfig);
-            config.put(FITBIT_USERS_CONFIG, u);
-            return config;
-          })
-          .collect(Collectors.toList());
-      this.configuredUsers = fitbitConfig.getUserRepository().stream()
-          .collect(Collectors.toSet());
-      logger.info("Received userTask Configs {}" , userTasks);
+    try {
+      for(UserRepository ur: fitbitConfig.getUserRepositories()) {
+        List<Map<String, String>> userTasksCurrent = ur.stream()
+            .map(User::getId)
+            // group users based on their hashCode
+            // in principle this allows for more efficient reconfigurations for a fixed number of tasks,
+            // since that allows existing tasks to only handle small modifications users to handle.
+            .collect(Collectors.groupingBy(
+                u -> Math.abs(u.hashCode()) % maxTasks,
+                Collectors.joining(",")))
+            .values().stream()
+            .map(u -> {
+              Map<String, String> config = new HashMap<>(baseConfig);
+              config.put(FITBIT_USERS_CONFIG, u);
+              return config;
+            })
+            .collect(Collectors.toList());
+        userTasks.addAll(userTasksCurrent);
+        Set currentUsers = ur.stream()
+            .collect(toSet());
+        this.configuredUsers.addAll(currentUsers);
+        logger.info("Received userTask Configs {}", userTasks);
+      }
       return userTasks;
     } catch (IOException ex) {
       throw new ConfigException("Cannot read users", ex);

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

@@ -23,12 +23,7 @@
 import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule;
 import io.confluent.connect.avro.AvroData;
 import java.io.IOException;
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
+import java.util.*;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 import okhttp3.OkHttpClient;
@@ -58,7 +53,7 @@ public class FitbitRequestGenerator extends RequestGeneratorRouter {
 
   private OkHttpClient baseClient;
   private final Map<String, OkHttpClient> clients;
-  private UserRepository userRepository;
+  private List<UserRepository> userRepositories;
   private List<RequestRoute> routes;
 
   public FitbitRequestGenerator() {
@@ -75,35 +70,42 @@ public void initialize(RestSourceConnectorConfig config) {
     FitbitRestSourceConnectorConfig fitbitConfig = (FitbitRestSourceConnectorConfig) config;
     this.baseClient = new OkHttpClient();
 
-    this.userRepository = fitbitConfig.getUserRepository();
+    this.userRepositories = fitbitConfig.getUserRepositories();
     this.routes = getRoutes(fitbitConfig);
 
     super.initialize(config);
   }
 
   private List<RequestRoute> getRoutes(FitbitRestSourceConnectorConfig config) {
     AvroData avroData = new AvroData(20);
-    List<RequestRoute> localRoutes = new ArrayList<>(5);
-    localRoutes.add(new FitbitSleepRoute(this, userRepository, avroData));
-    localRoutes.add(new FitbitTimeZoneRoute(this, userRepository, avroData));
-    localRoutes.add(new FitbitActivityLogRoute(this, userRepository, avroData));
-    if (config.hasIntradayAccess()) {
-      localRoutes.add(new FitbitIntradayStepsRoute(this, userRepository, avroData));
-      localRoutes.add(new FitbitIntradayHeartRateRoute(this, userRepository, avroData));
+    List<RequestRoute> localRoutes = new ArrayList<>(5 * userRepositories.size());
+    for(UserRepository userRepository : userRepositories) {
+      localRoutes.add(new FitbitSleepRoute(this, userRepository, avroData));
+      localRoutes.add(new FitbitTimeZoneRoute(this, userRepository, avroData));
+      localRoutes.add(new FitbitActivityLogRoute(this, userRepository, avroData));
+      if (config.hasIntradayAccess()) {
+        localRoutes.add(new FitbitIntradayStepsRoute(this, userRepository, avroData));
+        localRoutes.add(new FitbitIntradayHeartRateRoute(this, userRepository, avroData));
+      }
     }
     return localRoutes;
   }
 
-  public OkHttpClient getClient(User user) {
+  public OkHttpClient getClient(User user, UserRepository userRepository) {
+
     return clients.computeIfAbsent(user.getId(), u -> baseClient.newBuilder()
-          .authenticator(new TokenAuthenticator(user, userRepository))
-          .build());
+        .authenticator(new TokenAuthenticator(user, userRepository))
+        .build());
   }
 
   public Map<String, Map<String, Object>> getPartitions(String route) {
     try {
-      return userRepository.stream()
-          .collect(Collectors.toMap(User::getId, u -> getPartition(route, u)));
+      Map<String, Map<String, Object>> partitions = new HashMap<>();
+      for(UserRepository userRepository : userRepositories) {
+        partitions.putAll(userRepository.stream()
+            .collect(Collectors.toMap(User::getId, u -> getPartition(route, u))));
+      }
+      return partitions;
     } catch (IOException e) {
       logger.warn("Failed to initialize user partitions for route {}: {}", route, e.toString());
       return Collections.emptyMap();

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

@@ -31,12 +31,7 @@
 import java.time.ZonedDateTime;
 import java.time.format.DateTimeFormatter;
 import java.time.temporal.TemporalAmount;
-import java.util.AbstractMap;
-import java.util.Comparator;
-import java.util.HashMap;
-import java.util.Map;
-import java.util.Objects;
-import java.util.Set;
+import java.util.*;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
@@ -220,7 +215,7 @@ protected FitbitRestRequest newRequest(User user, DateRange dateRange,
           .header("Authorization", "Bearer " + userRepository.getAccessToken(user))
           .build();
       return new FitbitRestRequest(this, request, user, getPartition(user),
-          generator.getClient(user), dateRange,
+          generator.getClient(user, userRepository), dateRange,
           req -> !tooManyRequestsForUser.contains(((FitbitRestRequest)req).getUser()));
     } catch (NotAuthorizedException | IOException ex) {
       logger.warn("User {} does not have a configured access token: {}. Skipping.",