chore(*): Remove RxJava dependency from Igor (#778)

* chore(*): Remove RxJava dependency from Igor I removed the two tests that focused only on the scheduler. I'll argue that they actually test the framework and not the code, and because there's nothing equivalent to RxJava's `TestScheduler` in Spring, I figured I'd just delete the whole thing. * Add pool size property. Convert IgorConfig to Java. Co-authored-by: Gal Yardeni <55253849+gal-yardeni@users.noreply.github.com>
spinnaker · Jun 13, 2020 · af05e7f · af05e7f
1 parent eded1f4
commit af05e7f
Show file tree

Hide file tree

Showing 26 changed files with 249 additions and 534 deletions.
diff --git a/igor-core/igor-core.gradle b/igor-core/igor-core.gradle
@@ -4,9 +4,6 @@ dependencies {
   implementation "com.netflix.spinnaker.kork:kork-jedis"
   implementation "com.netflix.spinnaker.fiat:fiat-core:$fiatVersion"
 
-  // TODO(rz): Get rid of this dependency!
-  implementation "io.reactivex:rxjava"
-
   // TODO(rz): Get rid of this dependency!
   implementation "com.squareup.retrofit:retrofit"
 }
diff --git a/igor-core/src/main/java/com/netflix/spinnaker/igor/IgorConfigurationProperties.java b/igor-core/src/main/java/com/netflix/spinnaker/igor/IgorConfigurationProperties.java
@@ -75,6 +75,9 @@ public static class BuildProperties {
 
       /** TODO(jc): Please document */
       private boolean processBuildsOlderThanLookBackWindow = true;
+
+      /** The thread pool size used by the polling monitors */
+      private int schedulerPoolSize = Runtime.getRuntime().availableProcessors();
     }
 
     @Data

diff --git a/igor-core/src/main/java/com/netflix/spinnaker/igor/config/ArtifactTemplateProperties.java b/igor-core/src/main/java/com/netflix/spinnaker/igor/config/ArtifactTemplateProperties.java
@@ -22,9 +22,9 @@
 import java.util.Map;
 import java.util.stream.Collectors;
 import javax.validation.Valid;
+import javax.validation.constraints.NotEmpty;
 import lombok.Data;
 import lombok.NoArgsConstructor;
-import org.hibernate.validator.constraints.NotEmpty;
 import org.springframework.boot.context.properties.ConfigurationProperties;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.validation.annotation.Validated;

diff --git a/igor-core/src/main/java/com/netflix/spinnaker/igor/polling/CommonPollingMonitor.java b/igor-core/src/main/java/com/netflix/spinnaker/igor/polling/CommonPollingMonitor.java
@@ -26,15 +26,16 @@
 import com.netflix.spinnaker.kork.eureka.RemoteStatusChangedEvent;
 import java.time.Duration;
 import java.util.Optional;
+import java.util.concurrent.ScheduledFuture;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicLong;
 import javax.annotation.Nullable;
 import javax.annotation.PreDestroy;
 import net.logstash.logback.argument.StructuredArguments;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
-import rx.Scheduler;
-import rx.schedulers.Schedulers;
+import org.springframework.scheduling.TaskScheduler;
+import org.springframework.scheduling.support.PeriodicTrigger;
 
 public abstract class CommonPollingMonitor<I extends DeltaItem, T extends PollingDelta<I>>
     implements PollingMonitor, PollAccess {
@@ -49,38 +50,24 @@ public abstract class CommonPollingMonitor<I extends DeltaItem, T extends Pollin
   private final Id pollCycleFailedId;
   private final Id pollCycleTimingId;
   private final Optional<LockService> lockService;
+  private ScheduledFuture<?> monitor;
   protected Logger log = LoggerFactory.getLogger(getClass());
-  protected Scheduler.Worker worker;
+  protected TaskScheduler scheduler;
   protected final DynamicConfigService dynamicConfigService;
 
-  public CommonPollingMonitor(
-      IgorConfigurationProperties igorProperties,
-      Registry registry,
-      DynamicConfigService dynamicConfigService,
-      Optional<DiscoveryClient> discoveryClient,
-      Optional<LockService> lockService) {
-    this(
-        igorProperties,
-        registry,
-        dynamicConfigService,
-        discoveryClient,
-        lockService,
-        Schedulers.io());
-  }
-
   public CommonPollingMonitor(
       IgorConfigurationProperties igorProperties,
       Registry registry,
       DynamicConfigService dynamicConfigService,
       Optional<DiscoveryClient> discoveryClient,
       Optional<LockService> lockService,
-      Scheduler scheduler) {
+      TaskScheduler scheduler) {
     this.igorProperties = igorProperties;
     this.registry = registry;
     this.dynamicConfigService = dynamicConfigService;
     this.discoveryClient = discoveryClient;
     this.lockService = lockService;
-    this.worker = scheduler.createWorker();
+    this.scheduler = scheduler;
 
     itemsCachedId = registry.createId("pollingMonitor.newItems");
     itemsOverThresholdId = registry.createId("pollingMonitor.itemsOverThreshold");
@@ -97,32 +84,31 @@ public void onApplicationEvent(RemoteStatusChangedEvent event) {
     }
 
     initialize();
-    worker.schedulePeriodically(
-        () ->
-            registry
-                .timer(pollCycleTimingId.withTag("monitor", getClass().getSimpleName()))
-                .record(
-                    () -> {
-                      if (isInService()) {
-                        poll(true);
-                        lastPoll.set(System.currentTimeMillis());
-                      } else {
-                        log.info(
-                            "not in service (lastPoll: {})",
-                            (lastPoll == null) ? "n/a" : lastPoll.toString());
-                        lastPoll.set(0);
-                      }
-                    }),
-        0,
-        getPollInterval(),
-        TimeUnit.SECONDS);
+    this.monitor =
+        scheduler.schedule(
+            () ->
+                registry
+                    .timer(pollCycleTimingId.withTag("monitor", getClass().getSimpleName()))
+                    .record(
+                        () -> {
+                          if (isInService()) {
+                            poll(true);
+                            lastPoll.set(System.currentTimeMillis());
+                          } else {
+                            log.info(
+                                "not in service (lastPoll: {})",
+                                (lastPoll.get() == 0) ? "n/a" : lastPoll.toString());
+                            lastPoll.set(0);
+                          }
+                        }),
+            new PeriodicTrigger(getPollInterval(), TimeUnit.SECONDS));
   }
 
   @PreDestroy
   void stop() {
     log.info("Stopped");
-    if (!worker.isUnsubscribed()) {
-      worker.unsubscribe();
+    if (monitor != null && !monitor.isDone()) {
+      monitor.cancel(false);
     }
   }
 
@@ -284,8 +270,4 @@ private boolean isSendEventsEnabled() {
   protected @Nullable Integer getPartitionUpperThreshold(String partition) {
     return null;
   }
-
-  public void setWorker(Scheduler.Worker worker) {
-    this.worker = worker;
-  }
 }
diff --git a/igor-monitor-artifactory/igor-monitor-artifactory.gradle b/igor-monitor-artifactory/igor-monitor-artifactory.gradle
@@ -28,9 +28,6 @@ dependencies {
   // TODO(rz): Get rid of this dependency!
   implementation "com.netflix.spinnaker.kork:kork-jedis"
 
-  // TODO(rz): Get rid of this dependency!
-  implementation "io.reactivex:rxjava"
-
   // TODO(rz): Get rid of this dependency!
   implementation "net.logstash.logback:logstash-logback-encoder"
 

diff --git a/...factory/src/main/java/com/netflix/spinnaker/igor/artifactory/ArtifactoryBuildMonitor.java b/...factory/src/main/java/com/netflix/spinnaker/igor/artifactory/ArtifactoryBuildMonitor.java
@@ -54,6 +54,7 @@
 import org.jfrog.artifactory.client.aql.AqlQueryBuilder;
 import org.jfrog.artifactory.client.impl.ArtifactoryRequestImpl;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.scheduling.TaskScheduler;
 import org.springframework.stereotype.Service;
 
 @Service
@@ -74,8 +75,9 @@ public ArtifactoryBuildMonitor(
       Optional<LockService> lockService,
       Optional<EchoService> echoService,
       ArtifactoryCache cache,
-      ArtifactoryProperties artifactoryProperties) {
-    super(properties, registry, dynamicConfigService, discoveryClient, lockService);
+      ArtifactoryProperties artifactoryProperties,
+      TaskScheduler scheduler) {
+    super(properties, registry, dynamicConfigService, discoveryClient, lockService, scheduler);
     this.cache = cache;
     this.artifactoryProperties = artifactoryProperties;
     this.echoService = echoService;

diff --git a/...src/test/groovy/com/netflix/spinnaker/igor/artifactory/ArtifactoryBuildMonitorSpec.groovy b/...src/test/groovy/com/netflix/spinnaker/igor/artifactory/ArtifactoryBuildMonitorSpec.groovy
@@ -26,7 +26,7 @@ import com.netflix.spinnaker.igor.polling.LockService
 import com.netflix.spinnaker.kork.dynamicconfig.DynamicConfigService
 import com.squareup.okhttp.mockwebserver.MockResponse
 import com.squareup.okhttp.mockwebserver.MockWebServer
-import rx.schedulers.Schedulers
+import org.springframework.scheduling.TaskScheduler
 import spock.lang.Specification
 
 class ArtifactoryBuildMonitorSpec extends Specification {
@@ -47,9 +47,9 @@ class ArtifactoryBuildMonitorSpec extends Specification {
       Optional.ofNullable(lockService),
       Optional.of(echoService),
       cache,
-      new ArtifactoryProperties(searches: [search])
+      new ArtifactoryProperties(searches: [search]),
+      Mock(TaskScheduler)
     )
-    monitor.worker = Schedulers.immediate().createWorker()
 
     return monitor
   }

diff --git a/...monitor-plugins/src/main/java/com/netflix/spinnaker/igor/plugins/PluginsBuildMonitor.java b/...monitor-plugins/src/main/java/com/netflix/spinnaker/igor/plugins/PluginsBuildMonitor.java
@@ -23,7 +23,11 @@
 import com.netflix.spinnaker.igor.plugins.front50.PluginReleaseService;
 import com.netflix.spinnaker.igor.plugins.model.PluginEvent;
 import com.netflix.spinnaker.igor.plugins.model.PluginRelease;
-import com.netflix.spinnaker.igor.polling.*;
+import com.netflix.spinnaker.igor.polling.CommonPollingMonitor;
+import com.netflix.spinnaker.igor.polling.DeltaItem;
+import com.netflix.spinnaker.igor.polling.LockService;
+import com.netflix.spinnaker.igor.polling.PollContext;
+import com.netflix.spinnaker.igor.polling.PollingDelta;
 import com.netflix.spinnaker.kork.dynamicconfig.DynamicConfigService;
 import com.netflix.spinnaker.security.AuthenticatedRequest;
 import java.time.Instant;
@@ -32,6 +36,7 @@
 import java.util.Optional;
 import java.util.stream.Collectors;
 import lombok.Data;
+import org.springframework.scheduling.TaskScheduler;
 
 public class PluginsBuildMonitor
     extends CommonPollingMonitor<
@@ -49,8 +54,9 @@ public PluginsBuildMonitor(
       Optional<LockService> lockService,
       PluginReleaseService pluginInfoService,
       PluginCache cache,
-      Optional<EchoService> echoService) {
-    super(igorProperties, registry, dynamicConfigService, discoveryClient, lockService);
+      Optional<EchoService> echoService,
+      TaskScheduler scheduler) {
+    super(igorProperties, registry, dynamicConfigService, discoveryClient, lockService, scheduler);
     this.pluginInfoService = pluginInfoService;
     this.cache = cache;
     this.echoService = echoService;

diff --git a/igor-monitor-travis/src/main/java/com/netflix/spinnaker/igor/travis/TravisBuildMonitor.java b/igor-monitor-travis/src/main/java/com/netflix/spinnaker/igor/travis/TravisBuildMonitor.java
@@ -53,6 +53,7 @@
 import lombok.Getter;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.scheduling.TaskScheduler;
 import org.springframework.stereotype.Service;
 
 /** Monitors travis builds */
@@ -79,8 +80,9 @@ public TravisBuildMonitor(
       BuildServices buildServices,
       TravisProperties travisProperties,
       Optional<EchoService> echoService,
-      Optional<LockService> lockService) {
-    super(properties, registry, dynamicConfigService, discoveryClient, lockService);
+      Optional<LockService> lockService,
+      TaskScheduler scheduler) {
+    super(properties, registry, dynamicConfigService, discoveryClient, lockService, scheduler);
     this.buildCache = buildCache;
     this.buildServices = buildServices;
     this.travisProperties = travisProperties;

diff --git a/...or-travis/src/test/groovy/com/netflix/spinnaker/igor/travis/TravisBuildMonitorSpec.groovy b/...or-travis/src/test/groovy/com/netflix/spinnaker/igor/travis/TravisBuildMonitorSpec.groovy
@@ -30,6 +30,7 @@ import com.netflix.spinnaker.igor.travis.config.TravisProperties
 import com.netflix.spinnaker.igor.travis.service.TravisBuildConverter
 import com.netflix.spinnaker.igor.travis.service.TravisService
 import com.netflix.spinnaker.kork.dynamicconfig.DynamicConfigService
+import org.springframework.scheduling.TaskScheduler
 import spock.lang.Specification
 
 class TravisBuildMonitorSpec extends Specification {
@@ -55,7 +56,8 @@ class TravisBuildMonitorSpec extends Specification {
             buildServices,
             travisProperties,
             Optional.of(echoService),
-            Optional.empty()
+            Optional.empty(),
+            Mock(TaskScheduler)
         )
         travisService.isLogReady(_) >> true
         buildCache.getTrackedBuilds(MASTER) >> []

diff --git a/igor-web/igor-web.gradle b/igor-web/igor-web.gradle
@@ -79,7 +79,6 @@ dependencies {
     implementation "com.netflix.spinnaker.kork:kork-secrets-gcp"
     implementation "com.netflix.spinnaker.kork:kork-plugins"
 
-    implementation "io.reactivex:rxjava"
     implementation "com.netflix.eureka:eureka-client"
     implementation "io.github.resilience4j:resilience4j-retry"