BaragonRequestWorker implements Runnable { @Override public void run() { workerLastStartAt.set(System.currentTimeMillis()); try { final List<QueuedRequestWithState> queuedRequests = requestManager.getQueuedRequestIds() .stream() .map(this::hydrateQueuedRequestWithState) .filter(Optional::isPresent) .map(Optional::get) .collect(Collectors.toList()); List<QueuedRequestWithState> inFlightRequests = queuedRequests.stream() .filter((q) -> q.getCurrentState().isInFlight() || hasInProgressAttempt(q)) .collect(Collectors.toList()); final Set<String> inProgressServices = inFlightRequests.stream() .map((q) -> q.getQueuedRequestId().getServiceId()) .collect(Collectors.toSet()); final Set<QueuedRequestWithState> removedForCurrentInFlightRequest = queuedRequests.stream() .filter((q) -> inProgressServices.contains(q.getQueuedRequestId().getServiceId()) && !inFlightRequests.contains(q)) .collect(Collectors.toSet()); if (!inFlightRequests.isEmpty()) { LOG.info("Skipping new updates for services {} due to current in-flight updates", String.join(",", inProgressServices)); } queuedRequests.removeAll(removedForCurrentInFlightRequest); LOG.debug("Processing {} BaragonRequests which are already in-flight", inFlightRequests.size()); handleResultStates(handleQueuedRequests(inFlightRequests)); queuedRequests.removeAll(inFlightRequests); int added = inFlightRequests.size(); while (added < configuration.getWorkerConfiguration().getMaxRequestsPerPoll() && !queuedRequests.isEmpty()) { ArrayList<QueuedRequestWithState> nonServiceChanges = new ArrayList<>(); ArrayList<QueuedRequestWithState> serviceChanges = new ArrayList<>(); collectRequests(added, queuedRequests, nonServiceChanges, serviceChanges); List<QueuedRequestWithState> hydratedNonServiceChanges = nonServiceChanges.stream() .filter((q) -> { if (inProgressServices.contains(q.getQueuedRequestId().getServiceId())) { LOG.info("Skipping {} because {} already has an in progress request", q.getQueuedRequestId().getRequestId(), q.getQueuedRequestId().getServiceId()); return false; } return true; }) .map(someRequest -> new MaybeAdjustedRequest(someRequest, false)) .map(this::setNoValidateIfRequestRemovesUpstreamsOnly) .map(this::preResolveDNS) .map(this::saveAdjustedRequest) .sorted(queuedRequestComparator()) .collect(Collectors.toList()); added += hydratedNonServiceChanges.size(); LOG.debug("Processing {} BaragonRequests which don't modify a BaragonService", nonServiceChanges.size()); handleResultStates(handleQueuedRequests(hydratedNonServiceChanges)); queuedRequests.removeAll(nonServiceChanges); inProgressServices.addAll(hydratedNonServiceChanges.stream().map((q) -> q.getQueuedRequestId().getServiceId()).collect(Collectors.toSet())); List<QueuedRequestWithState> hydratedServiceChanges = serviceChanges.stream() .filter((q) -> { if (inProgressServices.contains(q.getQueuedRequestId().getServiceId())) { LOG.info("Skipping {} because {} already has an in progress request", q.getQueuedRequestId().getRequestId(), q.getQueuedRequestId().getServiceId()); return false; } return true; }) .sorted(queuedRequestComparator()) .collect(Collectors.toList()); added += hydratedServiceChanges.size(); LOG.debug("Processing {} BaragonRequests which modify a BaragonService", serviceChanges.size()); handleResultStates(handleQueuedRequests(hydratedServiceChanges)); queuedRequests.removeAll(serviceChanges); inProgressServices.addAll(hydratedServiceChanges.stream().map((q) -> q.getQueuedRequestId().getServiceId()).collect(Collectors.toSet())); } } catch (Exception e) { LOG.warn("Caught exception", e); exceptionNotifier.notify(e, Collections.emptyMap()); } finally { LOG.debug("Finished poller loop."); } } @Inject  BaragonRequestWorker(AgentManager agentManager,
                              RequestManager requestManager,
                              BaragonStateDatastore stateDatastore,
                              BaragonExceptionNotifier exceptionNotifier,
                              BaragonConfiguration configuration,
                              EdgeCache edgeCache,
                              UpstreamResolver resolver,
                              @Named(BaragonDataModule.BARAGON_SERVICE_WORKER_LAST_START) AtomicLong workerLastStartAt); @Override void run();  }
@Test public void testQueuedRequestsAreBatchedForAgent() throws Exception { String agentUrl = "http: startAgent(agentUrl, TEST_LB_GROUP); requestManager.enqueueRequest(createBaseRequest("request1", "service1", ImmutableSet.of(TEST_LB_GROUP)).build()); requestManager.enqueueRequest(createBaseRequest("request2", "service2", ImmutableSet.of(TEST_LB_GROUP)).build()); requestWorker.run(); requestWorker.run(); Assertions.assertEquals(2, testAgentManager.getRecentBatches().get(agentUrl).size()); }
@Test public void testServiceBatchBoundaryIsRespected() throws Exception { String agentUrl = "http: startAgent(agentUrl, TEST_LB_GROUP); requestManager.enqueueRequest(createBaseRequest("request1", "service1", ImmutableSet.of(TEST_LB_GROUP)).build()); requestManager.enqueueRequest(createBaseRequest("request2", "service1", ImmutableSet.of(TEST_LB_GROUP), ImmutableMap.of("1", "2")).build()); requestWorker.run(); requestWorker.run(); Assertions.assertEquals(1, testAgentManager.getRecentBatches().get(agentUrl).size()); }
@Test public void testInFlightRequestsAreRespected() throws Exception { String agentUrl = "http: startAgent(agentUrl, TEST_LB_GROUP); requestManager.enqueueRequest(createBaseRequest("request1", "service1", ImmutableSet.of(TEST_LB_GROUP)).build()); requestWorker.run(); requestWorker.run(); Assertions.assertEquals(1, testAgentManager.getRecentBatches().get(agentUrl).size()); requestManager.enqueueRequest(createBaseRequest("request2", "service1", ImmutableSet.of(TEST_LB_GROUP)).build()); Assertions.assertEquals(1, testAgentManager.getRecentBatches().get(agentUrl).size()); Assertions.assertEquals("request1", testAgentManager.getRecentBatches().get(agentUrl).get(0).getRequestId()); }
@Test public void testFailedRetriedRequestsAreConsideredInFlight() throws Exception { String agentUrl = "http: startAgent(agentUrl, TEST_LB_GROUP); requestManager.enqueueRequest(createBaseRequest("request1", "service1", ImmutableSet.of(TEST_LB_GROUP)).build()); requestWorker.run(); requestWorker.run(); Assertions.assertEquals(1, testAgentManager.getRecentBatches().get(agentUrl).size()); testAgentManager.completeRequestWithFailure(agentUrl, agentUrl, "request1", AgentRequestType.APPLY); requestWorker.run(); Assertions.assertEquals(InternalRequestStates.SEND_APPLY_REQUESTS, requestManager.getRequestState("request1").get()); requestManager.enqueueRequest(createBaseRequest("request2", "service1", ImmutableSet.of(TEST_LB_GROUP)).build()); Assertions.assertEquals(1, testAgentManager.getRecentBatches().get(agentUrl).size()); Assertions.assertEquals("request1", testAgentManager.getRecentBatches().get(agentUrl).get(0).getRequestId()); }