@Test public void testPlaceAllPendingPodsRegardlessOfBatchSize() { final DBConnectionPool dbConnectionPool = new DBConnectionPool(); final DSLContext conn = dbConnectionPool.getConnectionToDb(); final List<String> policies = Policies.getDefaultPolicies(); final NodeResourceEventHandler nodeResourceEventHandler = new NodeResourceEventHandler(dbConnectionPool); final int numNodes = 50; final int numPods = 102; conn.insertInto(Tables.BATCH_SIZE).values(numPods).execute(); final PodEventsToDatabase eventHandler = new PodEventsToDatabase(dbConnectionPool); final PodResourceEventHandler handler = new PodResourceEventHandler(eventHandler::handle); for (int i = 0; i < numNodes; i++) { nodeResourceEventHandler.onAddSync(newNode("n" + i, Collections.emptyMap(), Collections.emptyList())); final String podName = "system-pod-n" + i; final Pod pod = newPod(podName, "Running", Collections.emptyMap(), Collections.emptyMap()); pod.getSpec().setNodeName("n" + i); handler.onAddSync(pod); } for (int i = 0; i < numPods; i++) { handler.onAddSync(newPod("p" + i)); } final Scheduler scheduler = new Scheduler(dbConnectionPool, policies, "ORTOOLS", true, NUM_THREADS); scheduler.scheduleAllPendingPods(new EmulatedPodToNodeBinder(dbConnectionPool)); final Result<PodInfoRecord> fetch = conn.selectFrom(Tables.POD_INFO).fetch(); fetch.forEach(e -> assertTrue(e.getNodeName() != null && e.getNodeName().startsWith("n"))); }
@SuppressWarnings("unchecked") void scheduleAllPendingPods(final IPodToNodeBinder binder) { int fetchCount = dbConnectionPool.getConnectionToDb().fetchCount(Tables.PODS_TO_ASSIGN_NO_LIMIT); while (fetchCount > 0) { LOG.info("Fetchcount is {}", fetchCount); final int batch = batchId.incrementAndGet(); final long now = System.nanoTime(); final Result<? extends Record> podsToAssignUpdated = runOneLoop(); final long totalTime = System.nanoTime() - now; solverInvocations.mark(); fetchCount -= podsToAssignUpdated.size(); try (final DSLContext conn = dbConnectionPool.getConnectionToDb()) { final List<Update<?>> updates = new ArrayList<>(); podsToAssignUpdated.forEach(r -> { final String podName = r.get(Tables.PODS_TO_ASSIGN.POD_NAME); final String nodeName = r.get(Tables.PODS_TO_ASSIGN.CONTROLLABLE__NODE_NAME); updates.add( conn.update(Tables.POD_INFO) .set(Tables.POD_INFO.NODE_NAME, nodeName) .where(Tables.POD_INFO.POD_NAME.eq(podName)) ); LOG.info("Scheduling decision for pod {} as part of batch {} made in time: {}", podName, batch, totalTime); }); conn.batch(updates).execute(); } LOG.info("Done with updates"); binder.bindManyAsnc(podsToAssignUpdated); LOG.info("Done with bindings"); } }
Scheduler { @SuppressWarnings("unchecked") void scheduleAllPendingPods(final IPodToNodeBinder binder) { int fetchCount = dbConnectionPool.getConnectionToDb().fetchCount(Tables.PODS_TO_ASSIGN_NO_LIMIT); while (fetchCount > 0) { LOG.info("Fetchcount is {}", fetchCount); final int batch = batchId.incrementAndGet(); final long now = System.nanoTime(); final Result<? extends Record> podsToAssignUpdated = runOneLoop(); final long totalTime = System.nanoTime() - now; solverInvocations.mark(); fetchCount -= podsToAssignUpdated.size(); try (final DSLContext conn = dbConnectionPool.getConnectionToDb()) { final List<Update<?>> updates = new ArrayList<>(); podsToAssignUpdated.forEach(r -> { final String podName = r.get(Tables.PODS_TO_ASSIGN.POD_NAME); final String nodeName = r.get(Tables.PODS_TO_ASSIGN.CONTROLLABLE__NODE_NAME); updates.add( conn.update(Tables.POD_INFO) .set(Tables.POD_INFO.NODE_NAME, nodeName) .where(Tables.POD_INFO.POD_NAME.eq(podName)) ); LOG.info("Scheduling decision for pod {} as part of batch {} made in time: {}", podName, batch, totalTime); }); conn.batch(updates).execute(); } LOG.info("Done with updates"); binder.bindManyAsnc(podsToAssignUpdated); LOG.info("Done with bindings"); } } }
Scheduler { @SuppressWarnings("unchecked") void scheduleAllPendingPods(final IPodToNodeBinder binder) { int fetchCount = dbConnectionPool.getConnectionToDb().fetchCount(Tables.PODS_TO_ASSIGN_NO_LIMIT); while (fetchCount > 0) { LOG.info("Fetchcount is {}", fetchCount); final int batch = batchId.incrementAndGet(); final long now = System.nanoTime(); final Result<? extends Record> podsToAssignUpdated = runOneLoop(); final long totalTime = System.nanoTime() - now; solverInvocations.mark(); fetchCount -= podsToAssignUpdated.size(); try (final DSLContext conn = dbConnectionPool.getConnectionToDb()) { final List<Update<?>> updates = new ArrayList<>(); podsToAssignUpdated.forEach(r -> { final String podName = r.get(Tables.PODS_TO_ASSIGN.POD_NAME); final String nodeName = r.get(Tables.PODS_TO_ASSIGN.CONTROLLABLE__NODE_NAME); updates.add( conn.update(Tables.POD_INFO) .set(Tables.POD_INFO.NODE_NAME, nodeName) .where(Tables.POD_INFO.POD_NAME.eq(podName)) ); LOG.info("Scheduling decision for pod {} as part of batch {} made in time: {}", podName, batch, totalTime); }); conn.batch(updates).execute(); } LOG.info("Done with updates"); binder.bindManyAsnc(podsToAssignUpdated); LOG.info("Done with bindings"); } } Scheduler(final DBConnectionPool dbConnectionPool, final List<String> policies, final String solverToUse,
              final boolean debugMode, final int numThreads);   Scheduler(final DBConnectionPool dbConnectionPool, final List<String> policies, final String solverToUse,
              final boolean debugMode, final int numThreads, final int solverMaxTimeInSeconds); }
Scheduler { @SuppressWarnings("unchecked") void scheduleAllPendingPods(final IPodToNodeBinder binder) { int fetchCount = dbConnectionPool.getConnectionToDb().fetchCount(Tables.PODS_TO_ASSIGN_NO_LIMIT); while (fetchCount > 0) { LOG.info("Fetchcount is {}", fetchCount); final int batch = batchId.incrementAndGet(); final long now = System.nanoTime(); final Result<? extends Record> podsToAssignUpdated = runOneLoop(); final long totalTime = System.nanoTime() - now; solverInvocations.mark(); fetchCount -= podsToAssignUpdated.size(); try (final DSLContext conn = dbConnectionPool.getConnectionToDb()) { final List<Update<?>> updates = new ArrayList<>(); podsToAssignUpdated.forEach(r -> { final String podName = r.get(Tables.PODS_TO_ASSIGN.POD_NAME); final String nodeName = r.get(Tables.PODS_TO_ASSIGN.CONTROLLABLE__NODE_NAME); updates.add( conn.update(Tables.POD_INFO) .set(Tables.POD_INFO.NODE_NAME, nodeName) .where(Tables.POD_INFO.POD_NAME.eq(podName)) ); LOG.info("Scheduling decision for pod {} as part of batch {} made in time: {}", podName, batch, totalTime); }); conn.batch(updates).execute(); } LOG.info("Done with updates"); binder.bindManyAsnc(podsToAssignUpdated); LOG.info("Done with bindings"); } } Scheduler(final DBConnectionPool dbConnectionPool, final List<String> policies, final String solverToUse,
              final boolean debugMode, final int numThreads);   Scheduler(final DBConnectionPool dbConnectionPool, final List<String> policies, final String solverToUse,
              final boolean debugMode, final int numThreads, final int solverMaxTimeInSeconds); static void main(final String[] args); }
Scheduler { @SuppressWarnings("unchecked") void scheduleAllPendingPods(final IPodToNodeBinder binder) { int fetchCount = dbConnectionPool.getConnectionToDb().fetchCount(Tables.PODS_TO_ASSIGN_NO_LIMIT); while (fetchCount > 0) { LOG.info("Fetchcount is {}", fetchCount); final int batch = batchId.incrementAndGet(); final long now = System.nanoTime(); final Result<? extends Record> podsToAssignUpdated = runOneLoop(); final long totalTime = System.nanoTime() - now; solverInvocations.mark(); fetchCount -= podsToAssignUpdated.size(); try (final DSLContext conn = dbConnectionPool.getConnectionToDb()) { final List<Update<?>> updates = new ArrayList<>(); podsToAssignUpdated.forEach(r -> { final String podName = r.get(Tables.PODS_TO_ASSIGN.POD_NAME); final String nodeName = r.get(Tables.PODS_TO_ASSIGN.CONTROLLABLE__NODE_NAME); updates.add( conn.update(Tables.POD_INFO) .set(Tables.POD_INFO.NODE_NAME, nodeName) .where(Tables.POD_INFO.POD_NAME.eq(podName)) ); LOG.info("Scheduling decision for pod {} as part of batch {} made in time: {}", podName, batch, totalTime); }); conn.batch(updates).execute(); } LOG.info("Done with updates"); binder.bindManyAsnc(podsToAssignUpdated); LOG.info("Done with bindings"); } } Scheduler(final DBConnectionPool dbConnectionPool, final List<String> policies, final String solverToUse,
              final boolean debugMode, final int numThreads);   Scheduler(final DBConnectionPool dbConnectionPool, final List<String> policies, final String solverToUse,
              final boolean debugMode, final int numThreads, final int solverMaxTimeInSeconds); static void main(final String[] args);  }