Skip to content

improve: PerResourcePollingEventSource init improvements #1918

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 5 commits into from
May 30, 2023
Merged
Show file tree
Hide file tree
Changes from 2 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
@@ -1,5 +1,7 @@
package io.javaoperatorsdk.operator.processing.dependent.external;

import java.time.Duration;

import io.fabric8.kubernetes.api.model.HasMetadata;
import io.javaoperatorsdk.operator.api.reconciler.EventSourceContext;
import io.javaoperatorsdk.operator.api.reconciler.Ignore;
Expand All @@ -23,8 +25,8 @@ public PerResourcePollingDependentResource(Class<R> resourceType, long pollingPe
@Override
protected ExternalResourceCachingEventSource<R, P> createEventSource(
EventSourceContext<P> context) {
return new PerResourcePollingEventSource<>(this, context.getPrimaryCache(),
getPollingPeriod(), resourceType(), this);
return new PerResourcePollingEventSource<>(this, context,
Duration.ofMillis(getPollingPeriod()), resourceType(), this);
}

}
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,7 @@

import io.fabric8.kubernetes.api.model.HasMetadata;
import io.javaoperatorsdk.operator.OperatorException;
import io.javaoperatorsdk.operator.api.reconciler.EventSourceContext;
import io.javaoperatorsdk.operator.processing.event.ResourceID;
import io.javaoperatorsdk.operator.processing.event.source.Cache;
import io.javaoperatorsdk.operator.processing.event.source.CacheKeyMapper;
Expand Down Expand Up @@ -43,19 +44,49 @@ public class PerResourcePollingEventSource<R, P extends HasMetadata>
private final long period;
private final Set<ResourceID> fetchedForPrimaries = ConcurrentHashMap.newKeySet();

public PerResourcePollingEventSource(ResourceFetcher<R, P> resourceFetcher,
EventSourceContext<P> context, Duration defaultPollingPeriod,
Class<R> resourceClass) {
this(resourceFetcher, context.getPrimaryCache(), defaultPollingPeriod.toMillis(),
null, resourceClass,
CacheKeyMapper.singleResourceCacheKeyMapper());
}

@Deprecated(forRemoval = true)
Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Specify in the javadoc using the @deprecated attribute what is supposed to be used instead.

public PerResourcePollingEventSource(ResourceFetcher<R, P> resourceFetcher,
Cache<P> resourceCache, long period, Class<R> resourceClass) {
this(resourceFetcher, resourceCache, period, null, resourceClass,
CacheKeyMapper.singleResourceCacheKeyMapper());
}

public PerResourcePollingEventSource(ResourceFetcher<R, P> resourceFetcher,
EventSourceContext<P> context,
Duration defaultPollingPeriod,
Class<R> resourceClass,
CacheKeyMapper<R> cacheKeyMapper) {
this(resourceFetcher, context.getPrimaryCache(), defaultPollingPeriod.toMillis(),
null, resourceClass, cacheKeyMapper);
}

@Deprecated(forRemoval = true)
Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Same as above.

public PerResourcePollingEventSource(ResourceFetcher<R, P> resourceFetcher,
Cache<P> resourceCache, long period, Class<R> resourceClass,
CacheKeyMapper<R> cacheKeyMapper) {
this(resourceFetcher, resourceCache, period, null, resourceClass, cacheKeyMapper);
}

public PerResourcePollingEventSource(ResourceFetcher<R, P> resourceFetcher,
EventSourceContext<P> context,
Duration defaultPollingPeriod,
Predicate<P> registerPredicate,
Class<R> resourceClass,
CacheKeyMapper<R> cacheKeyMapper) {
this(resourceFetcher, context.getPrimaryCache(), defaultPollingPeriod.toMillis(),
registerPredicate, resourceClass, cacheKeyMapper,
new ScheduledThreadPoolExecutor(DEFAULT_EXECUTOR_THREAD_NUMBER));
}

@Deprecated(forRemoval = true)
Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Same as above

public PerResourcePollingEventSource(ResourceFetcher<R, P> resourceFetcher,
Cache<P> resourceCache, long period,
Predicate<P> registerPredicate, Class<R> resourceClass,
Expand All @@ -64,7 +95,20 @@ public PerResourcePollingEventSource(ResourceFetcher<R, P> resourceFetcher,
new ScheduledThreadPoolExecutor(DEFAULT_EXECUTOR_THREAD_NUMBER));
}

public PerResourcePollingEventSource(ResourceFetcher<R, P> resourceFetcher,

public PerResourcePollingEventSource(
ResourceFetcher<R, P> resourceFetcher,
EventSourceContext<P> context, Duration defaultPollingPeriod,
Predicate<P> registerPredicate, Class<R> resourceClass,
CacheKeyMapper<R> cacheKeyMapper, ScheduledExecutorService executorService) {
this(resourceFetcher, context.getPrimaryCache(), defaultPollingPeriod.toMillis(),
registerPredicate,
resourceClass, cacheKeyMapper, executorService);
}

@Deprecated
Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Same as above.

public PerResourcePollingEventSource(
ResourceFetcher<R, P> resourceFetcher,
Cache<P> resourceCache, long period,
Predicate<P> registerPredicate, Class<R> resourceClass,
CacheKeyMapper<R> cacheKeyMapper, ScheduledExecutorService executorService) {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -9,11 +9,9 @@
import org.junit.jupiter.api.Test;

import io.javaoperatorsdk.operator.TestUtils;
import io.javaoperatorsdk.operator.api.reconciler.EventSourceContext;
import io.javaoperatorsdk.operator.processing.event.EventHandler;
import io.javaoperatorsdk.operator.processing.event.source.AbstractEventSourceTestBase;
import io.javaoperatorsdk.operator.processing.event.source.Cache;
import io.javaoperatorsdk.operator.processing.event.source.CacheKeyMapper;
import io.javaoperatorsdk.operator.processing.event.source.SampleExternalResource;
import io.javaoperatorsdk.operator.processing.event.source.*;
import io.javaoperatorsdk.operator.sample.simple.TestCustomResource;

import static org.assertj.core.api.Assertions.assertThat;
Expand All @@ -35,16 +33,19 @@ class PerResourcePollingEventSourceTest extends
private final PerResourcePollingEventSource.ResourceFetcher<SampleExternalResource, TestCustomResource> supplier =
mock(PerResourcePollingEventSource.ResourceFetcher.class);
@SuppressWarnings("unchecked")
private final Cache<TestCustomResource> resourceCache = mock(Cache.class);
private final IndexerResourceCache<TestCustomResource> resourceCache =
mock(IndexerResourceCache.class);
private final TestCustomResource testCustomResource = TestUtils.testCustomResource();
private final EventSourceContext<TestCustomResource> context = mock(EventSourceContext.class);

@BeforeEach
public void setup() {
when(resourceCache.get(any())).thenReturn(Optional.of(testCustomResource));
when(supplier.fetchResources(any()))
.thenReturn(Set.of(SampleExternalResource.testResource1()));
when(context.getPrimaryCache()).thenReturn(resourceCache);

setUpSource(new PerResourcePollingEventSource<>(supplier, resourceCache, PERIOD,
setUpSource(new PerResourcePollingEventSource<>(supplier, context, Duration.ofMillis(PERIOD),
SampleExternalResource.class, r -> r.getName() + "#" + r.getValue()));
}

Expand All @@ -61,7 +62,7 @@ void pollsTheResourceAfterAwareOfIt() {

@Test
void registeringTaskOnAPredicate() {
setUpSource(new PerResourcePollingEventSource<>(supplier, resourceCache, PERIOD,
setUpSource(new PerResourcePollingEventSource<>(supplier, context, Duration.ofMillis(PERIOD),
testCustomResource -> testCustomResource.getMetadata().getGeneration() > 1,
SampleExternalResource.class, CacheKeyMapper.singleResourceCacheKeyMapper()));
source.onResourceCreated(testCustomResource);
Expand Down
Original file line number Diff line number Diff line change
@@ -1,5 +1,6 @@
package io.javaoperatorsdk.operator.sample.externalstate;

import java.time.Duration;
import java.util.Collections;
import java.util.Map;
import java.util.Set;
Expand Down Expand Up @@ -122,7 +123,7 @@ public Map<String, EventSource> prepareEventSources(
var id = configMap.getData().get(ID_KEY);
var externalResource = externalService.read(id);
return externalResource.map(Set::of).orElseGet(Collections::emptySet);
}, context.getPrimaryCache(), 300L, ExternalResource.class);
}, context, Duration.ofMillis(300L), ExternalResource.class);

return EventSourceInitializer.nameEventSources(configMapEventSource,
externalResourceEventSource);
Expand Down
Original file line number Diff line number Diff line change
@@ -1,5 +1,6 @@
package io.javaoperatorsdk.operator.sample.perresourceeventsource;

import java.time.Duration;
import java.util.Map;
import java.util.Set;
import java.util.UUID;
Expand Down Expand Up @@ -41,7 +42,7 @@ public Map<String, EventSource> prepareEventSources(
numberOfFetchExecutions.compute(resource.getMetadata().getName(), (s, v) -> v + 1);
return Set.of(UUID.randomUUID().toString());
},
context.getPrimaryCache(), POLL_PERIOD, String.class);
context, Duration.ofMillis(POLL_PERIOD), String.class);
return EventSourceInitializer.nameEventSources(eventSource);
}

Expand Down