feat: Dynamic agent registeration & multiple stomp topic & scheduler feature

bm-agent/build.gradle

@@ -7,20 +7,32 @@ plugins {
 }
 
 group = 'org.benchmarker'
-version = '0.0.1-SNAPSHOT'
+version = '1.0.0'
 
 def excludeJacocoTestCoverageReport = [
         // bmagent
         'org/benchmarker/bmagent/BmAgentApplication**',
         'org/benchmarker/bmagent/sse/**',
         'org/benchmarker/bmagent/consts/SystemSchedulerConst',
+        'org/benchmarker/bmagent/status/**',
+        'org/benchmarker/bmagent/initializer/**',
 //        'org/benchmarker/bmagent/**',
 ]
 
 java {
     sourceCompatibility = '17'
 }
 
+ext {
+    set('springCloudVersion', "2023.0.0")
+}
+
+dependencyManagement {
+    imports {
+        mavenBom "org.springframework.cloud:spring-cloud-dependencies:${springCloudVersion}"
+    }
+}
+
 jacoco {
     toolVersion = "0.8.8"
 }
@@ -41,16 +53,23 @@ dependencies {
     implementation 'org.projectlombok:lombok'
     annotationProcessor 'org.projectlombok:lombok'
     implementation 'com.fasterxml.jackson.datatype:jackson-datatype-jsr310'
+    implementation 'org.springframework.boot:spring-boot-starter-actuator'
     // add webflux
     implementation 'org.springframework.boot:spring-boot-starter-webflux'
     testImplementation 'org.springframework.boot:spring-boot-starter-test'
     testImplementation 'org.springframework.boot:spring-boot-testcontainers'
     testImplementation 'org.testcontainers:junit-jupiter'
     testImplementation 'com.squareup.okhttp3:mockwebserver:4.9.1'
+
+    implementation "io.jsonwebtoken:jjwt-api:0.11.1"
+    runtimeOnly "io.jsonwebtoken:jjwt-impl:0.11.1"
+    runtimeOnly "io.jsonwebtoken:jjwt-jackson:0.11.1"
     if (isAppleSilicon()) {
         runtimeOnly("io.netty:netty-resolver-dns-native-macos:4.1.94.Final:osx-aarch_64")
     }
 
+    implementation 'org.springframework.cloud:spring-cloud-starter-netflix-eureka-client'
+
     implementation project(':bm-common')
 }
 

bm-agent/src/main/java/org/benchmarker/bmagent/BmAgentApplication.java

@@ -1,13 +1,24 @@
 package org.benchmarker.bmagent;
 
+import org.springframework.boot.ExitCodeGenerator;
 import org.springframework.boot.SpringApplication;
 import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.context.annotation.Bean;
 
 @SpringBootApplication
 public class BmAgentApplication {
+    @Bean
+    public ExitCodeGenerator exitCodeGenerator() {
+        return new ExitCodeGenerator() {
+            @Override
+            public int getExitCode() {
+                return 42;
+            }
+        };
+    }
 
     public static void main(String[] args) {
+        System.setProperty("spring.application.name","bm-agent");
         SpringApplication.run(BmAgentApplication.class, args);
     }
-
 }

bm-agent/src/main/java/org/benchmarker/bmagent/consts/HeaderConst.java

@@ -0,0 +1,7 @@
+package org.benchmarker.bmagent.consts;
+
+public interface HeaderConst {
+    String bearerPrefix = "Bearer ";
+    String tokenKey = "Authorization";
+
+}

bm-agent/src/main/java/org/benchmarker/bmagent/consts/SystemSchedulerConst.java

@@ -7,5 +7,8 @@ public interface SystemSchedulerConst {
      */
     Long systemSchedulerId = -100L;
     String systemUsageSchedulerName = "cpu-memory-usage-update";
+    Integer connectControllerTimeout = 10; // seconds
+    Integer connectionFailedLimit = 50;
+
 
 }

bm-agent/src/main/java/org/benchmarker/bmagent/controller/AgentApiController.java

@@ -3,6 +3,7 @@
 
 import jakarta.servlet.http.HttpServletRequest;
 import java.util.Map;
+import java.util.Set;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.benchmarker.bmagent.AgentInfo;
@@ -42,15 +43,16 @@ public class AgentApiController {
      * @param action     String
      * @return SseEmitter
      */
-    @PostMapping("/templates/{template_id}")
+    @PostMapping("/groups/{group_id}/templates/{template_id}")
     public SseEmitter manageSSE(@PathVariable("template_id") Long templateId,
+        @PathVariable("group_id") String groupId,
         @RequestParam("action") String action, @RequestBody TemplateInfo templateInfo) {
         log.info(templateInfo.toString());
-        agentStatusManager.getAndUpdateStatusIfReady(
-            AgentStatus.TESTING).orElseThrow(() -> new RuntimeException("agent is not ready"));
 
         if (action.equals("start")) {
-            return sseManageService.start(templateId, templateInfo);
+            agentStatusManager.getAndUpdateStatusIfReady(
+                AgentStatus.TESTING).orElseThrow(() -> new RuntimeException("agent is not ready"));
+            return sseManageService.start(templateId, groupId, templateInfo);
         } else {
             sseManageService.stop(templateId);
             return null;
@@ -73,10 +75,12 @@ public AgentInfo getStatus() {
         String scheme = request.getScheme(); // http or https
         String serverName = request.getServerName();
         int serverPort = request.getServerPort();
+        Set<Long> longs = scheduledTaskService.getStatus().keySet();
 
         String agentServerUrl = scheme + "://" + serverName + ":" + serverPort;
 
         return AgentInfo.builder()
+            .templateId(longs)
             .cpuUsage(agentStatusManager.getCpuUsage())
             .memoryUsage(agentStatusManager.getMemoryUsage())
             .startedAt(agentStatusManager.getStartedAt())

bm-agent/src/main/java/org/benchmarker/bmagent/initializer/Initializer.java

@@ -2,24 +2,39 @@
 
 import java.util.concurrent.TimeUnit;
 import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
 import org.benchmarker.bmagent.consts.SystemSchedulerConst;
 import org.benchmarker.bmagent.schedule.ScheduledTaskService;
 import org.benchmarker.bmagent.status.AgentStatusManager;
+import org.springframework.beans.BeansException;
 import org.springframework.boot.CommandLineRunner;
+import org.springframework.context.ApplicationContext;
+import org.springframework.context.ApplicationContextAware;
+import org.springframework.context.annotation.Profile;
 import org.springframework.stereotype.Component;
 
 @Component
+@Profile("!test")
+@Slf4j
 @RequiredArgsConstructor
-public class Initializer implements CommandLineRunner {
+public class Initializer implements CommandLineRunner, ApplicationContextAware {
 
     private final ScheduledTaskService scheduledTaskService;
     private final AgentStatusManager agentStatusManager;
+    private ApplicationContext applicationContext;
+
 
     @Override
     public void run(String... args) throws Exception {
+        log.info("init");
         // cpu, memory usage checker
         scheduledTaskService.startChild(SystemSchedulerConst.systemSchedulerId,
             SystemSchedulerConst.systemUsageSchedulerName, agentStatusManager::updateStats, 0, 1,
             TimeUnit.SECONDS);
     }
+
+    @Override
+    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
+        this.applicationContext = applicationContext;
+    }
 }

bm-agent/src/main/java/org/benchmarker/bmagent/pref/HttpSender.java

@@ -17,10 +17,13 @@
 import java.util.stream.IntStream;
 import lombok.Getter;
 import lombok.extern.slf4j.Slf4j;
+import org.benchmarker.bmagent.AgentStatus;
 import org.benchmarker.bmagent.service.IScheduledTaskService;
+import org.benchmarker.bmagent.status.AgentStatusManager;
 import org.benchmarker.bmagent.util.WebClientSupport;
 import org.benchmarker.bmcommon.dto.TemplateInfo;
 import org.springframework.web.reactive.function.client.WebClient.RequestHeadersSpec;
+import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
 import reactor.core.publisher.Mono;
 
 /**
@@ -35,14 +38,17 @@ public class HttpSender {
     private final ResultManagerService resultManagerService;
     private final IScheduledTaskService scheduledTaskService;
 
+    private final AgentStatusManager agentStatusManager;
+
 
     public HttpSender(ResultManagerService resultManagerService,
-        IScheduledTaskService scheduledTaskService) {
+        IScheduledTaskService scheduledTaskService, AgentStatusManager agentStatusManager) {
         this.resultManagerService = resultManagerService;
         this.scheduledTaskService = scheduledTaskService;
+        this.agentStatusManager = agentStatusManager;
     }
 
-    private Integer defaultMaxRequestsPerUser = 100000000;
+    private Integer defaultMaxRequestsPerUser = Integer.MAX_VALUE;
     private Integer defaultMaxDuration = 5; // 5 hours
     private AtomicInteger totalRequests = new AtomicInteger(0);
     private AtomicInteger totalSuccess = new AtomicInteger(0);
@@ -66,7 +72,8 @@ public HttpSender(ResultManagerService resultManagerService,
      *
      * @param templateInfo {@link TemplateInfo}
      */
-    public void sendRequests(TemplateInfo templateInfo) throws MalformedURLException {
+    public void sendRequests(SseEmitter sseEmitter, TemplateInfo templateInfo) throws MalformedURLException {
+
         URL url = new URL(templateInfo.getUrl());
         RequestHeadersSpec<?> req = WebClientSupport.create(templateInfo.getMethod(),
             templateInfo.getUrl(),
@@ -85,16 +92,20 @@ public void sendRequests(TemplateInfo templateInfo) throws MalformedURLException
         }
 
         Duration duration = templateInfo.getMaxDuration();
+        log.info("Now send multiple HTTP request to target server");
+        log.info(templateInfo.toString());
 
         // Future setup
         futures = IntStream.range(0, templateInfo.getVuser())
             .mapToObj(i -> CompletableFuture.runAsync(() -> {
                 long startTime = System.currentTimeMillis(); // 시작 시간 기록
                 long endTime = startTime + duration.toMillis();
+
                 for (int j = 0; j < templateInfo.getMaxRequest(); j++) {
                     // 만약 running 이 아니거나 시간이 끝났다면,
                     if (!isRunning || System.currentTimeMillis() > endTime) {
-                        return;
+                        agentStatusManager.updateAgentStatus(AgentStatus.READY);
+                        break;
                     }
                     long requestStartTime = System.currentTimeMillis();  // 요청 시작 시간 기록
                     req.exchangeToMono(resp -> {
@@ -192,6 +203,7 @@ public Map<Double, Double> calculateMttfbPercentile(List<Double> percentile) {
      * downstream 에서만 제거 가능. 외부에서는 cancel 불가능...
      */
     public void cancelRequests() {
+        log.info("cancel requests");
         isRunning = false;
 
     }

bm-agent/src/main/java/org/benchmarker/bmagent/security/JwtTokenProvider.java

@@ -0,0 +1,79 @@
+package org.benchmarker.bmagent.security;
+
+
+import io.jsonwebtoken.Claims;
+import io.jsonwebtoken.Jwts;
+import io.jsonwebtoken.SignatureAlgorithm;
+import java.util.Date;
+import lombok.Setter;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.stereotype.Component;
+
+/**
+ * {@link JwtTokenProvider} class is used to create Json Web Token.
+ */
+@Slf4j
+@Component
+@Setter
+public class JwtTokenProvider {
+
+    /**
+     * Access token's expiration time
+     */
+    public static String expirationTime;
+
+    /**
+     * Refresh token's expiration time
+     */
+    public static String refreshExpirationTime;
+
+    /**
+     * Secret key for signing the token
+     */
+    public static String secret;
+
+    @Value("${token.expiration_time}")
+    private void setExpirationTime(String expirationTime) {
+        JwtTokenProvider.expirationTime = expirationTime;
+    }
+
+    @Value("${token.refresh_expiration_time}")
+    private void setRefreshExpirationTime(String refreshExpirationTime) {
+        JwtTokenProvider.refreshExpirationTime = refreshExpirationTime;
+    }
+
+    @Value("${token.secret}")
+    private void setSecret(String secret) {
+        JwtTokenProvider.secret = secret;
+    }
+
+
+    /**
+     * Create Json Web Token with user's username and authorities
+     *
+     * <p> Here, this method generate accessToken </p>
+     * <p> JWT's payload will looks like below</p>
+     * <blockquote><pre>
+     *  {
+     *      "sub": "agent"
+     *  }
+     * </pre></blockquote>
+     *
+     * @return Json-web-token
+     */
+    public static String createAccessToken() {
+        Claims claims = Jwts.claims().setSubject("agent");
+
+        Long expirationTimeLong = Long.parseLong(expirationTime);
+        final Date createdDate = new Date();
+        final Date expirationDate = new Date(createdDate.getTime() + expirationTimeLong);
+        return Jwts.builder()
+            .setClaims(claims)
+            .setSubject("agent")
+            .setIssuedAt(createdDate)
+            .setExpiration(expirationDate)
+            .signWith(SignatureAlgorithm.HS512, secret)
+            .compact();
+    }
+}

bm-agent/src/main/java/org/benchmarker/bmagent/service/ISseManageService.java

@@ -19,7 +19,7 @@ public interface ISseManageService extends SseManageConsts {
      * @param templateInfo TemplateInfo
      * @return SseEmitter
      */
-    SseEmitter start(Long id, TemplateInfo templateInfo);
+    SseEmitter start(Long id, String groupId, TemplateInfo templateInfo);
 
     /**
      * Stop the SSE emitter for the given id