Skip to content

Commit

Permalink
refactor(core): Remove RxJava from ExecutionRepository interface (spi…
Browse files Browse the repository at this point in the history 
…nnaker#2205)

* refactor(core): retrieve to not use RxJava in return type

* refactor(core): retrievePipelinesForApplication to not use RxJava in return type

* refactor(core): retrievePipelinesForPipelineConfigId to not use RxJava in return type

* refactor(core): retrieveOrchestrationsForApplication to not use RxJava in return type

* refactor(core): retrievePipelinesForApplication to not use RxJava in return type

* refactor(core): retrievePipelinesForPipelineConfigId to not use RxJava in return type

* Merge changes around pending execution
  • Loading branch information
@robzienert
robzienert authored May 8, 2018
1 parent 2a34fa8 commit 313460a
Show file tree
Hide file tree
Showing 18 changed files with 187 additions and 187 deletions.
18 changes: 9 additions & 9 deletions ...main/groovy/com/netflix/spinnaker/orca/pipeline/persistence/ExecutionRepositoryTck.groovy
View file
Original file line number Diff line number Diff line change
Expand Up @@ -22,10 +22,10 @@ import com.netflix.spinnaker.orca.pipeline.model.Execution
import com.netflix.spinnaker.orca.pipeline.model.JenkinsTrigger
import com.netflix.spinnaker.orca.pipeline.model.PipelineTrigger
import com.netflix.spinnaker.orca.pipeline.persistence.ExecutionRepository.ExecutionCriteria
import rx.schedulers.Schedulers
import spock.lang.Specification
import spock.lang.Subject
import spock.lang.Unroll

import static com.netflix.spinnaker.orca.ExecutionStatus.*
import static com.netflix.spinnaker.orca.pipeline.model.Execution.ExecutionType
import static com.netflix.spinnaker.orca.pipeline.model.Execution.ExecutionType.PIPELINE
Expand Down Expand Up @@ -66,23 +66,23 @@ abstract class ExecutionRepositoryTck<T extends ExecutionRepository> extends Spe
repository.store(succeededExecution)
def pipelines = repository.retrievePipelinesForPipelineConfigId(
"pipeline-1", new ExecutionCriteria(limit: 5, statuses: ["RUNNING", "SUCCEEDED", "TERMINAL"])
).subscribeOn(Schedulers.io()).toList().toBlocking().single()
).toList()

then:
pipelines*.id.sort() == [runningExecution.id, succeededExecution.id].sort()

when:
pipelines = repository.retrievePipelinesForPipelineConfigId(
"pipeline-1", new ExecutionCriteria(limit: 5, statuses: ["RUNNING"])
).subscribeOn(Schedulers.io()).toList().toBlocking().single()
).toList()

then:
pipelines*.id.sort() == [runningExecution.id].sort()

when:
pipelines = repository.retrievePipelinesForPipelineConfigId(
"pipeline-1", new ExecutionCriteria(limit: 5, statuses: ["TERMINAL"])
).subscribeOn(Schedulers.io()).toList().toBlocking().single()
).toList()

then:
pipelines.isEmpty()
Expand All @@ -106,23 +106,23 @@ abstract class ExecutionRepositoryTck<T extends ExecutionRepository> extends Spe
repository.store(succeededExecution)
def orchestrations = repository.retrieveOrchestrationsForApplication(
runningExecution.application, new ExecutionCriteria(limit: 5, statuses: ["RUNNING", "SUCCEEDED", "TERMINAL"])
).subscribeOn(Schedulers.io()).toList().toBlocking().single()
).toList()

then:
orchestrations*.id.sort() == [runningExecution.id, succeededExecution.id].sort()

when:
orchestrations = repository.retrieveOrchestrationsForApplication(
runningExecution.application, new ExecutionCriteria(limit: 5, statuses: ["RUNNING"])
).subscribeOn(Schedulers.io()).toList().toBlocking().single()
).toList()

then:
orchestrations*.id.sort() == [runningExecution.id].sort()

when:
orchestrations = repository.retrieveOrchestrationsForApplication(
runningExecution.application, new ExecutionCriteria(limit: 5, statuses: ["TERMINAL"])
).subscribeOn(Schedulers.io()).toList().toBlocking().single()
).toList()

then:
orchestrations.isEmpty()
Expand Down Expand Up @@ -154,7 +154,7 @@ abstract class ExecutionRepositoryTck<T extends ExecutionRepository> extends Spe
repository.store(pipeline)

expect:
repository.retrieve(PIPELINE).toBlocking().first().id == pipeline.id
repository.retrieve(PIPELINE).first().id == pipeline.id

with(repository.retrieve(pipeline.type, pipeline.id)) {
id == pipeline.id
Expand Down Expand Up @@ -223,7 +223,7 @@ abstract class ExecutionRepositoryTck<T extends ExecutionRepository> extends Spe
thrown ExecutionNotFoundException

and:
repository.retrieve(PIPELINE).toList().toBlocking().first() == []
repository.retrieve(PIPELINE).toList() == []
}

def "updateStatus sets startTime to current time if new status is RUNNING"() {
Expand Down
18 changes: 13 additions & 5 deletions ...re/src/main/java/com/netflix/spinnaker/orca/pipeline/persistence/ExecutionRepository.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -19,11 +19,12 @@
import com.netflix.spinnaker.orca.pipeline.model.Execution;
import com.netflix.spinnaker.orca.pipeline.model.Execution.ExecutionType;
import com.netflix.spinnaker.orca.pipeline.model.Stage;
import rx.Observable;

import javax.annotation.Nonnull;
import javax.annotation.Nullable;
import java.util.*;
import java.util.stream.Stream;
import java.util.stream.StreamSupport;

import static java.util.stream.Collectors.toList;

Expand Down Expand Up @@ -56,14 +57,14 @@ public interface ExecutionRepository {

void delete(@Nonnull ExecutionType type, @Nonnull String id);

@Nonnull Observable<Execution> retrieve(ExecutionType type);
@Nonnull Iterable<Execution> retrieve(ExecutionType type);

@Nonnull Observable<Execution> retrievePipelinesForApplication(@Nonnull String application);
@Nonnull Iterable<Execution> retrievePipelinesForApplication(@Nonnull String application);

@Nonnull Observable<Execution> retrievePipelinesForPipelineConfigId(
@Nonnull Iterable<Execution> retrievePipelinesForPipelineConfigId(
@Nonnull String pipelineConfigId, @Nonnull ExecutionCriteria criteria);

@Nonnull Observable<Execution> retrieveOrchestrationsForApplication(
@Nonnull Iterable<Execution> retrieveOrchestrationsForApplication(
@Nonnull String application, @Nonnull ExecutionCriteria criteria);

@Nonnull Execution retrieveOrchestrationForCorrelationId(
Expand Down Expand Up @@ -116,4 +117,11 @@ public ExecutionCriteria setStatuses(ExecutionStatus... statuses) {
return Objects.hash(limit, statuses);
}
}

final class IterableUtil {

public static <T> Stream<T> toStream(Iterable<T> iterable) {
return StreamSupport.stream(Spliterators.spliteratorUnknownSize(iterable.iterator(), Spliterator.ORDERED), false);
}
}
}
27 changes: 7 additions & 20 deletions orca-core/src/main/java/com/netflix/spinnaker/orca/pipeline/util/ArtifactResolver.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -30,23 +30,15 @@
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import rx.functions.Func2;
import rx.schedulers.Schedulers;

import javax.annotation.Nonnull;
import javax.annotation.Nullable;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;
import java.util.Set;
import java.util.*;
import java.util.stream.Collectors;
import java.util.stream.Stream;

import static com.netflix.spinnaker.orca.pipeline.persistence.ExecutionRepository.IterableUtil.toStream;
import static java.lang.String.format;
import static java.util.Collections.emptyList;
import static java.util.stream.Collectors.toList;
Expand Down Expand Up @@ -139,15 +131,10 @@ List<Artifact> getArtifactsForPipelineId(
@Nonnull String pipelineId,
@Nonnull ExecutionCriteria criteria
) {
Execution execution = executionRepository
.retrievePipelinesForPipelineConfigId(pipelineId, criteria)
.subscribeOn(Schedulers.io())
.toSortedList(startTimeOrId)
.toBlocking()
.single()
.stream()
.findFirst()
.orElse(null);
Execution execution = toStream(executionRepository.retrievePipelinesForPipelineConfigId(pipelineId, criteria))
.sorted(startTimeOrId)
.findFirst()
.orElse(null);

return execution == null ? Collections.emptyList() : getAllArtifacts(execution);
}
Expand Down Expand Up @@ -266,7 +253,7 @@ public static class ResolveResult {
Set<ExpectedArtifact> unresolvedExpectedArtifacts = new HashSet<>();
}

private static Func2<Execution, Execution, Integer> startTimeOrId = (a, b) -> {
private static Comparator<Execution> startTimeOrId = (a, b) -> {
Long aStartTime = Optional.ofNullable(a.getStartTime()).orElse(0L);
Long bStartTime = Optional.ofNullable(b.getStartTime()).orElse(0L);

Expand Down
24 changes: 12 additions & 12 deletions ...te/src/main/java/com/netflix/spinnaker/orca/pipelinetemplate/PipelineTemplateService.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -15,10 +15,6 @@
*/
package com.netflix.spinnaker.orca.pipelinetemplate;

import java.util.List;
import java.util.Map;
import java.util.NoSuchElementException;
import javax.annotation.Nullable;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.netflix.spinnaker.orca.jackson.OrcaObjectMapper;
import com.netflix.spinnaker.orca.pipeline.model.Execution;
Expand All @@ -33,6 +29,13 @@
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.annotation.Nullable;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.NoSuchElementException;

import static com.netflix.spinnaker.orca.pipeline.model.Execution.ExecutionType.PIPELINE;

@Component
Expand Down Expand Up @@ -85,15 +88,12 @@ public Execution retrievePipelineOrNewestExecution(@Nullable String executionId,
} else if (pipelineConfigId != null) {
// No executionId set - use last execution
ExecutionRepository.ExecutionCriteria criteria = new ExecutionRepository.ExecutionCriteria().setLimit(1);
try {
return executionRepository.retrievePipelinesForPipelineConfigId(pipelineConfigId, criteria)
.toSingle()
.toBlocking()
.value();
} catch (NoSuchElementException e) {
throw new ExecutionNotFoundException("No pipeline execution could be found for config id " +
pipelineConfigId + ": " + e.getMessage());

Iterator<Execution> executions = executionRepository.retrievePipelinesForPipelineConfigId(pipelineConfigId, criteria).iterator();
if (!executions.hasNext()) {
throw new ExecutionNotFoundException("No pipeline execution could be found for config id " + pipelineConfigId);
}
return executions.next();
} else {
throw new IllegalArgumentException("Either executionId or pipelineConfigId have to be set.");
}
Expand Down
5 changes: 2 additions & 3 deletions orca-queue/src/main/kotlin/com/netflix/spinnaker/orca/q/handler/OrcaMessageHandler.kt
View file
Original file line number Diff line number Diff line change
Expand Up @@ -96,9 +96,8 @@ internal interface OrcaMessageHandler<M : Message> : MessageHandler<M> {
val criteria = ExecutionRepository.ExecutionCriteria().setLimit(1).setStatuses(ExecutionStatus.RUNNING)
!repository
.retrievePipelinesForPipelineConfigId(configId, criteria)
.isEmpty
.toBlocking()
.first()
.iterator()
.hasNext()
} == true
}
}
11 changes: 2 additions & 9 deletions orca-queue/src/test/kotlin/com/netflix/spinnaker/orca/q/handler/StartExecutionHandlerTest.kt
View file
Original file line number Diff line number Diff line change
Expand Up @@ -38,7 +38,6 @@ import org.jetbrains.spek.api.dsl.*
import org.jetbrains.spek.api.lifecycle.CachingMode.GROUP
import org.jetbrains.spek.subject.SubjectSpek
import org.springframework.context.ApplicationEventPublisher
import rx.Observable.just
import java.util.*

object StartExecutionHandlerTest : SubjectSpek<StartExecutionHandler>({
Expand Down Expand Up @@ -276,10 +275,7 @@ object StartExecutionHandlerTest : SubjectSpek<StartExecutionHandler>({
pipeline.isLimitConcurrent = true
runningPipeline.isLimitConcurrent = true

whenever(
repository
.retrievePipelinesForPipelineConfigId(configId, ExecutionCriteria().setLimit(1).setStatuses(RUNNING))
) doReturn just(runningPipeline)
whenever(pendingExecutionService.depth(configId)) doReturn 1
whenever(
repository.retrieve(message.executionType, message.executionId)
) doReturn pipeline
Expand Down Expand Up @@ -310,10 +306,7 @@ object StartExecutionHandlerTest : SubjectSpek<StartExecutionHandler>({
pipeline.isLimitConcurrent = false
runningPipeline.isLimitConcurrent = false

whenever(
repository
.retrievePipelinesForPipelineConfigId(configId, ExecutionCriteria().setLimit(1).setStatuses(RUNNING))
) doReturn just(runningPipeline)
whenever(pendingExecutionService.depth(configId)) doReturn 1
whenever(
repository.retrieve(message.executionType, message.executionId)
) doReturn pipeline
Expand Down
4 changes: 1 addition & 3 deletions ...etflix/spinnaker/orca/migrations/MultiRedisOrchestrationMigrationNotificationAgent.groovy
View file
Original file line number Diff line number Diff line change
Expand Up @@ -122,10 +122,8 @@ class MultiRedisOrchestrationMigrationNotificationAgent extends AbstractPollingN
def applicationName = application.name.toLowerCase()
def unmigratedOrchestrations = executionRepositoryPrevious
.retrieveOrchestrationsForApplication(applicationName, executionCriteria)
.filter({ orchestration -> !previouslyMigratedOrchestrationIds.contains(orchestration.id) })
.toList()
.toBlocking()
.single()
.findAll { orchestration -> !previouslyMigratedOrchestrationIds.contains(orchestration.id) }

def migratableOrchestrations = unmigratedOrchestrations.findAll { it.status.isComplete() }
def pendingOrchestrations = unmigratedOrchestrations.findAll { !it.status.isComplete() }
Expand Down
4 changes: 1 addition & 3 deletions ...com/netflix/spinnaker/orca/migrations/MultiRedisPipelineMigrationNotificationAgent.groovy
View file
Original file line number Diff line number Diff line change
Expand Up @@ -120,10 +120,8 @@ class MultiRedisPipelineMigrationNotificationAgent extends AbstractPollingNotifi
allPipelineConfigIds.eachWithIndex { String pipelineConfigId, int index ->
def unmigratedPipelines = executionRepositoryPrevious
.retrievePipelinesForPipelineConfigId(pipelineConfigId, executionCriteria)
.filter({ pipeline -> !previouslyMigratedPipelineIds.contains(pipeline.id) })
.toList()
.toBlocking()
.single()
.findAll { pipeline -> !previouslyMigratedPipelineIds.contains(pipeline.id) }

def migratablePipelines = unmigratedPipelines.findAll { it.status.isComplete() }
def pendingPipelines = unmigratedPipelines.findAll { !it.status.isComplete() }
Expand Down
17 changes: 10 additions & 7 deletions ...x/spinnaker/orca/notifications/scheduling/OldPipelineCleanupPollingNotificationAgent.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -31,7 +31,6 @@
import rx.Observable;
import rx.Scheduler;
import rx.Subscription;
import rx.functions.Func1;
import rx.schedulers.Schedulers;

import javax.annotation.PreDestroy;
Expand All @@ -40,9 +39,13 @@
import java.time.temporal.ChronoUnit;
import java.util.*;
import java.util.concurrent.TimeUnit;
import java.util.function.Function;
import java.util.function.Predicate;
import java.util.stream.Collectors;
import java.util.stream.Stream;

import static com.netflix.spinnaker.orca.pipeline.model.Execution.ExecutionType.PIPELINE;
import static com.netflix.spinnaker.orca.pipeline.persistence.ExecutionRepository.IterableUtil.toStream;

// TODO rz - Remove redis know-how and move back into orca-core
@Component
Expand All @@ -56,9 +59,9 @@ public class OldPipelineCleanupPollingNotificationAgent implements ApplicationLi
private Scheduler scheduler = Schedulers.io();
private Subscription subscription;

private Func1<Execution, Boolean> filter = new Func1<Execution, Boolean>() {
private Predicate<Execution> filter = new Predicate<Execution>() {
@Override
public Boolean call(Execution execution) {
public boolean test(Execution execution) {
if (!COMPLETED_STATUSES.contains(execution.getStatus().toString())) {
return false;
}
Expand All @@ -67,7 +70,7 @@ public Boolean call(Execution execution) {
}
};

private Func1<Execution, PipelineExecutionDetails> mapper = execution -> new PipelineExecutionDetails(
private Function<Execution, PipelineExecutionDetails> mapper = execution -> new PipelineExecutionDetails(
execution.getId(),
execution.getApplication(),
execution.getPipelineConfigId() == null ? "ungrouped" : execution.getPipelineConfigId(),
Expand Down Expand Up @@ -138,16 +141,16 @@ private void tick() {

applications.forEach(app -> {
log.debug("Cleaning up " + app);
cleanupApp(executionRepository.retrievePipelinesForApplication(app));
cleanupApp(toStream(executionRepository.retrievePipelinesForApplication(app)));
});

} catch (Exception e) {
log.error("Cleanup failed", e);
}
}

private void cleanupApp(Observable<Execution> observable) {
List<PipelineExecutionDetails> allPipelines = observable.filter(filter).map(mapper).toList().toBlocking().single();
private void cleanupApp(Stream<Execution> executions) {
List<PipelineExecutionDetails> allPipelines = executions.filter(filter).map(mapper).collect(Collectors.toList());

Map<String, List<PipelineExecutionDetails>> groupedPipelines = new HashMap<>();
allPipelines.forEach(p -> {
Expand Down
Loading

0 comments on commit 313460a

Please sign in to comment.