Merge pull request #45 from digma-ai/feature/rds-load

Feature/rds load

Author: ohrebeniuk

src/main/java/org/springframework/samples/petclinic/clinicactivity/ClinicActivityController.java

@@ -161,6 +161,39 @@ public ResponseEntity<String> recreateAndPopulateLogs(@RequestParam(name = "coun
 		}
 	}
 
+	@PostMapping("/io-intensive-load")
+	public ResponseEntity<String> createIOIntensiveLoad(@RequestParam(name = "duration", defaultValue = "5") int durationMinutes,
+														@RequestParam(name = "threads", defaultValue = "6") int numThreads,
+														@RequestParam(name = "limit", defaultValue = "400000") int limit) {
+		logger.warn("Received request to create I/O INTENSIVE LOAD for {} minutes with {} threads and {} limit - This will MAX OUT disk I/O operations!",
+					durationMinutes, numThreads, limit);
+		if (durationMinutes <= 0) {
+			return ResponseEntity.badRequest().body("Duration must be a positive integer.");
+		}
+		if (durationMinutes > 60) {
+			return ResponseEntity.badRequest().body("Duration too high for I/O intensive load - maximum 60 minutes to prevent storage overload.");
+		}
+		if (numThreads <= 0) {
+			return ResponseEntity.badRequest().body("Number of threads must be a positive integer.");
+		}
+		if (numThreads > 20) {
+			return ResponseEntity.badRequest().body("Too many threads for I/O intensive load - maximum 20 to prevent system crash.");
+		}
+		if (limit <= 0) {
+			return ResponseEntity.badRequest().body("Limit must be a positive integer.");
+		}
+		if (limit > 1000000) {
+			return ResponseEntity.badRequest().body("Limit too high for I/O intensive load - maximum 1,000,000 to prevent excessive resource usage.");
+		}
+		try {
+			dataService.createIOIntensiveLoad(durationMinutes, numThreads, limit);
+			return ResponseEntity.ok("Successfully completed I/O INTENSIVE LOAD for " + durationMinutes + " minutes with " + numThreads + " threads and " + limit + " limit - Disk I/O was maxed out!");
+		} catch (Exception e) {
+			logger.error("Error during I/O intensive load", e);
+			return ResponseEntity.status(HttpStatus.INTERNAL_SERVER_ERROR).body("Error during I/O intensive load: " + e.getMessage());
+		}
+	}
+
     private void performObservableOperation(String operationName) {
         Span span = otelTracer.spanBuilder(operationName)
             .setSpanKind(SpanKind.CLIENT)

src/main/java/org/springframework/samples/petclinic/clinicactivity/ClinicActivityDataService.java

@@ -25,7 +25,13 @@
 import java.util.List;
 import java.util.Locale;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.atomic.AtomicInteger;
 import java.util.Random;
+import java.util.Map;
+import java.util.HashMap;
 
 @Service
 public class ClinicActivityDataService {
@@ -47,6 +53,7 @@ public class ClinicActivityDataService {
             "Emergency Alert", "Consultation Note", "Follow-up Reminder"
     );
     private final Random random = new Random();
+    private final ExecutorService executorService = Executors.newFixedThreadPool(8);
 
     @Autowired
     public ClinicActivityDataService(ClinicActivityLogRepository repository,
@@ -200,4 +207,106 @@ private String csv(String value) {
         String escaped = value.replace("\"", "\"\"").replace("\\", "\\\\");
         return '"' + escaped + '"';
     }
+
+    public void createIOIntensiveLoad(int durationMinutes, int numThreads, int limit) {
+        logger.warn("Starting I/O INTENSIVE load test for {} minutes with {} threads and {} limit - This will MAX OUT disk I/O operations!",
+                   durationMinutes, numThreads, limit);
+        long startTime = System.currentTimeMillis();
+        long endTime = startTime + (durationMinutes * 60 * 1000L);
+
+        try {
+            AtomicInteger globalOperationCount = new AtomicInteger(0);
+            List<Thread> threads = new ArrayList<>();
+
+            logger.info("Creating {} I/O intensive threads with {} record limit per query...", numThreads, limit);
+
+            // Create I/O intensive threads
+            for (int t = 0; t < numThreads; t++) {
+                final int threadId = t;
+                Thread ioThread = new Thread(() -> {
+                    try {
+                        executeIOIntensiveThread(threadId, endTime, globalOperationCount, limit);
+                    } catch (Exception e) {
+                        logger.error("Error in I/O intensive thread {}", threadId, e);
+                    }
+                });
+
+                ioThread.setName("IOIntensiveThread-" + threadId);
+                threads.add(ioThread);
+            }
+
+            // Start all threads
+            logger.info("Starting all {} I/O intensive threads...", numThreads);
+            for (Thread thread : threads) {
+                thread.start();
+            }
+
+            // Wait for all threads to complete
+            for (Thread thread : threads) {
+                try {
+                    thread.join();
+                } catch (InterruptedException e) {
+                    Thread.currentThread().interrupt();
+                    logger.warn("Interrupted while waiting for I/O thread: {}", thread.getName());
+                }
+            }
+
+            long actualEndTime = System.currentTimeMillis();
+            logger.warn("Completed I/O INTENSIVE load test in {} ms with {} threads and {} limit. Total operations: {}",
+                (actualEndTime - startTime), numThreads, limit, globalOperationCount.get());
+
+        } catch (Exception e) {
+            logger.error("Error during I/O intensive load test", e);
+            throw new RuntimeException("Error during I/O intensive load test: " + e.getMessage(), e);
+        }
+    }
+
+    private void executeIOIntensiveThread(int threadId, long endTime, AtomicInteger globalOperationCount, int limit) {
+        Random random = new Random();
+        Faker faker = new Faker(new Locale("en-US"));
+        int localOperationCount = 0;
+
+        logger.info("I/O Thread {} starting I/O intensive operations with {} record limit...", threadId, limit);
+
+        while (System.currentTimeMillis() < endTime) {
+			try {
+                        // LARGE SEQUENTIAL SCAN - Forces full table scan I/O
+                        jdbcTemplate.queryForList(
+                            "SET work_mem = '512MB';" +
+								"SELECT id, activity_type, numeric_value, event_timestamp, payload " +
+                            "FROM clinic_activity_logs " +
+                            "WHERE LENGTH(payload) > 100 " +
+							"ORDER BY random()" +
+                            "LIMIT " + limit);
+
+
+                localOperationCount++;
+                int currentGlobalCount = globalOperationCount.incrementAndGet();
+
+                // Log progress every 100 operations per thread
+                if (localOperationCount % 100 == 0) {
+                    long remainingTime = (endTime - System.currentTimeMillis()) / 1000;
+                    logger.info("I/O Thread {} completed {} operations (Global: {}). Time remaining: {}s",
+                        threadId, localOperationCount, currentGlobalCount, remainingTime);
+                }
+
+                // No sleep - continuous I/O operations for maximum I/O pressure
+                // But avoid overwhelming the system with a tiny yield
+                if (localOperationCount % 50 == 0) {
+                    Thread.yield();
+                }
+
+            } catch (Exception e) {
+                logger.error("Error in I/O operation for thread {}", threadId, e);
+                try {
+                    Thread.sleep(10); // Brief pause on error
+                } catch (InterruptedException ie) {
+                    Thread.currentThread().interrupt();
+                    break;
+                }
+            }
+        }
+
+        logger.info("I/O Thread {} completed {} total I/O operations", threadId, localOperationCount);
+    }
 }