[Studio] Harden data source management

server/src/main/java/com/rocketmq/studio/settings/DataSourceTestDTO.java

@@ -16,6 +16,7 @@
  */
 package com.rocketmq.studio.settings;
 
+import jakarta.validation.constraints.NotBlank;
 import lombok.AllArgsConstructor;
 import lombok.Builder;
 import lombok.Data;
@@ -26,7 +27,12 @@
 @NoArgsConstructor
 @AllArgsConstructor
 public class DataSourceTestDTO {
+    @NotBlank(message = "url is required")
     private String url;
+    @NotBlank(message = "type is required")
     private String type;
     private String auth;
+    private String username;
+    private String password;
+    private String bearerToken;
 }

server/src/main/java/com/rocketmq/studio/settings/DataSourceVO.java

@@ -16,6 +16,7 @@
  */
 package com.rocketmq.studio.settings;
 
+import jakarta.validation.constraints.NotBlank;
 import lombok.AllArgsConstructor;
 import lombok.Builder;
 import lombok.Data;
@@ -27,8 +28,11 @@
 @AllArgsConstructor
 public class DataSourceVO {
     private String key;
+    @NotBlank(message = "name is required")
     private String name;
+    @NotBlank(message = "type is required")
     private String type;
+    @NotBlank(message = "url is required")
     private String url;
     private String auth;
     private String status;

server/src/main/java/com/rocketmq/studio/settings/SettingsController.java

@@ -52,12 +52,12 @@ public Result<List<DataSourceVO>> listDataSources() {
     }
 
     @PostMapping("/datasources/create")
-    public Result<DataSourceVO> createDataSource(@RequestBody DataSourceVO dataSource) {
+    public Result<DataSourceVO> createDataSource(@Valid @RequestBody DataSourceVO dataSource) {
         return Result.ok(settingsService.createDataSource(dataSource));
     }
 
     @PostMapping("/datasources/update")
-    public Result<DataSourceVO> updateDataSource(@RequestBody DataSourceVO dataSource) {
+    public Result<DataSourceVO> updateDataSource(@Valid @RequestBody DataSourceVO dataSource) {
         return Result.ok(settingsService.updateDataSource(dataSource));
     }
 
@@ -68,7 +68,7 @@ public Result<Void> deleteDataSource(@RequestParam String key) {
     }
 
     @PostMapping("/datasources/test")
-    public Result<DataSourceTestResultVO> testDataSource(@RequestBody DataSourceTestDTO request) {
+    public Result<DataSourceTestResultVO> testDataSource(@Valid @RequestBody DataSourceTestDTO request) {
         return Result.ok(settingsService.testDataSource(request));
     }
 }

server/src/main/java/com/rocketmq/studio/settings/SettingsService.java

@@ -16,20 +16,55 @@
  */
 package com.rocketmq.studio.settings;
 
-import lombok.RequiredArgsConstructor;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
 import lombok.extern.slf4j.Slf4j;
+import org.springframework.http.HttpHeaders;
+import org.springframework.http.MediaType;
+import org.springframework.http.client.SimpleClientHttpRequestFactory;
 import org.springframework.stereotype.Service;
 import org.springframework.util.StringUtils;
-
+import org.springframework.web.client.ResourceAccessException;
+import org.springframework.web.client.RestClient;
+import org.springframework.web.client.RestClientException;
+import org.springframework.web.client.RestClientResponseException;
+import org.springframework.web.util.UriComponentsBuilder;
+
+import java.io.IOException;
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.net.SocketTimeoutException;
+import java.time.Duration;
 import java.util.List;
+import java.util.Set;
 import java.util.UUID;
 
 @Slf4j
 @Service
-@RequiredArgsConstructor
 public class SettingsService {
 
+    private static final Set<String> PROMETHEUS_COMPATIBLE_TYPES = Set.of(
+            "prometheus", "victoriametrics", "thanos", "mimir");
+    private static final String PROMETHEUS_TEST_QUERY = "up";
+    private static final String AUTH_NONE = "none";
+    private static final String AUTH_BASIC = "basic auth";
+    private static final String AUTH_BEARER = "bearer token";
+    private static final Duration DATA_SOURCE_TEST_CONNECT_TIMEOUT = Duration.ofSeconds(3);
+    private static final Duration DATA_SOURCE_TEST_READ_TIMEOUT = Duration.ofSeconds(5);
+
     private final SettingsRepository settingsRepository;
+    private final RestClient restClient;
+    private final ObjectMapper objectMapper;
+
+    public SettingsService(SettingsRepository settingsRepository, RestClient.Builder restClientBuilder,
+                           ObjectMapper objectMapper) {
+        SimpleClientHttpRequestFactory requestFactory = new SimpleClientHttpRequestFactory();
+        requestFactory.setConnectTimeout(DATA_SOURCE_TEST_CONNECT_TIMEOUT);
+        requestFactory.setReadTimeout(DATA_SOURCE_TEST_READ_TIMEOUT);
+        this.settingsRepository = settingsRepository;
+        this.restClient = restClientBuilder.requestFactory(requestFactory).build();
+        this.objectMapper = objectMapper;
+    }
 
 
     public GeneralSettingsVO getGeneralSettings() {
@@ -77,11 +112,136 @@ public void deleteDataSource(String key) {
 
 
     public DataSourceTestResultVO testDataSource(DataSourceTestDTO request) {
-        log.info("Testing data source connection: url={}, type={}", request.getUrl(), request.getType());
-        // Stub: always return success for now
+        log.info("Testing data source connection: type={}", request == null ? null : request.getType());
+        if (request == null) {
+            return failed("Data source test request is required");
+        }
+        if (!isPrometheusCompatible(request.getType())) {
+            return failed("Unsupported data source type: " + request.getType());
+        }
+
+        try {
+            JsonNode response = restClient.get()
+                    .uri(prometheusQueryUri(request.getUrl()))
+                    .accept(MediaType.APPLICATION_JSON)
+                    .headers(headers -> applyAuthentication(headers, request))
+                    .retrieve()
+                    .body(JsonNode.class);
+            return prometheusSuccess(response);
+        } catch (IllegalArgumentException | URISyntaxException exception) {
+            return failed(exception.getMessage());
+        } catch (RestClientResponseException exception) {
+            return failed(prometheusErrorMessage(exception));
+        } catch (ResourceAccessException exception) {
+            if (hasCause(exception, SocketTimeoutException.class)) {
+                return failed("Prometheus connection timed out");
+            }
+            return failed("Failed to connect to Prometheus");
+        } catch (RestClientException exception) {
+            if (hasCause(exception, SocketTimeoutException.class)) {
+                return failed("Prometheus connection timed out");
+            }
+            return failed("Prometheus connection failed");
+        }
+    }
+
+    private boolean isPrometheusCompatible(String type) {
+        return StringUtils.hasText(type)
+                && PROMETHEUS_COMPATIBLE_TYPES.contains(type.replaceAll("\\s+", "").toLowerCase());
+    }
+
+    private void applyAuthentication(HttpHeaders headers, DataSourceTestDTO request) {
+        String auth = normalizeAuth(request.getAuth());
+        if (AUTH_NONE.equals(auth)) {
+            return;
+        }
+        if (AUTH_BASIC.equals(auth)) {
+            if (!StringUtils.hasText(request.getUsername()) || !StringUtils.hasText(request.getPassword())) {
+                throw new IllegalArgumentException("Basic authentication requires username and password");
+            }
+            headers.setBasicAuth(request.getUsername().trim(), request.getPassword());
+            return;
+        }
+        if (AUTH_BEARER.equals(auth)) {
+            if (!StringUtils.hasText(request.getBearerToken())) {
+                throw new IllegalArgumentException("Bearer authentication requires token");
+            }
+            headers.setBearerAuth(request.getBearerToken().trim());
+            return;
+        }
+        throw new IllegalArgumentException("Unsupported data source authentication: " + request.getAuth());
+    }
+
+    private String normalizeAuth(String auth) {
+        if (!StringUtils.hasText(auth)) {
+            return AUTH_NONE;
+        }
+        return auth.trim().replaceAll("\\s+", " ").toLowerCase();
+    }
+
+    private URI prometheusQueryUri(String baseUrl) throws URISyntaxException {
+        if (!StringUtils.hasText(baseUrl)) {
+            throw new IllegalArgumentException("Data source URL is required");
+        }
+        String normalized = baseUrl.strip();
+        while (normalized.endsWith("/")) {
+            normalized = normalized.substring(0, normalized.length() - 1);
+        }
+
+        URI baseUri = new URI(normalized);
+        if (!"http".equalsIgnoreCase(baseUri.getScheme()) && !"https".equalsIgnoreCase(baseUri.getScheme())) {
+            throw new IllegalArgumentException("Data source URL must start with http:// or https://");
+        }
+        return UriComponentsBuilder.fromUriString(normalized + "/api/v1/query")
+                .queryParam("query", PROMETHEUS_TEST_QUERY)
+                .build()
+                .toUri();
+    }
+
+    private DataSourceTestResultVO prometheusSuccess(JsonNode response) {
+        if (response != null && "success".equals(response.path("status").asText())) {
+            return DataSourceTestResultVO.builder()
+                    .success(true)
+                    .message("Connection successful")
+                    .build();
+        }
+        return failed(prometheusBodyError(response));
+    }
+
+    private String prometheusErrorMessage(RestClientResponseException exception) {
+        try {
+            return prometheusBodyError(objectMapper.readTree(exception.getResponseBodyAsString()));
+        } catch (IOException ignored) {
+            return "Prometheus query failed";
+        }
+    }
+
+    private String prometheusBodyError(JsonNode response) {
+        String errorType = response == null ? "" : response.path("errorType").asText();
+        String error = response == null ? "" : response.path("error").asText();
+        if (StringUtils.hasText(error)) {
+            return StringUtils.hasText(errorType)
+                    ? "Prometheus query failed (" + errorType + "): " + error
+                    : "Prometheus query failed: " + error;
+        }
+        return "Prometheus query failed";
+    }
+
+    private DataSourceTestResultVO failed(String message) {
         return DataSourceTestResultVO.builder()
-                .success(true)
-                .message("Connection successful")
+                .success(false)
+                .message(message)
                 .build();
     }
+
+    private boolean hasCause(Throwable throwable, Class<? extends Throwable> causeType) {
+        Throwable current = throwable;
+        while (current != null) {
+            if (causeType.isInstance(current)) {
+                return true;
+            }
+            current = current.getCause();
+        }
+        return false;
+    }
 }

server/src/test/java/com/rocketmq/studio/settings/SettingsControllerTest.java

@@ -203,6 +203,23 @@ void createDataSourceShouldReturnCreatedSource() throws Exception {
                 .andExpect(jsonPath("$.data.status", is("connected")));
     }
 
+    @Test
+    void createDataSourceShouldRejectMissingUrl() throws Exception {
+        mockMvc.perform(post("/api/settings/datasources/create")
+                        .contentType(MediaType.APPLICATION_JSON)
+                        .content("""
+                                {
+                                  "name": "New DS",
+                                  "type": "prometheus"
+                                }
+                                """))
+                .andExpect(status().isBadRequest())
+                .andExpect(jsonPath("$.code", is(400)))
+                .andExpect(jsonPath("$.message", is("url is required")));
+
+        verifyNoInteractions(settingsService);
+    }
+
     @Test
     void updateDataSourceShouldReturnUpdatedSource() throws Exception {
         DataSourceVO input = DataSourceVO.builder().key("ds-1").name("Updated DS").type("rocketmq")
@@ -218,6 +235,24 @@ void updateDataSourceShouldReturnUpdatedSource() throws Exception {
                 .andExpect(jsonPath("$.data.name", is("Updated DS")));
     }
 
+    @Test
+    void updateDataSourceShouldRejectMissingName() throws Exception {
+        mockMvc.perform(post("/api/settings/datasources/update")
+                        .contentType(MediaType.APPLICATION_JSON)
+                        .content("""
+                                {
+                                  "key": "ds-1",
+                                  "type": "rocketmq",
+                                  "url": "updated:9876"
+                                }
+                                """))
+                .andExpect(status().isBadRequest())
+                .andExpect(jsonPath("$.code", is(400)))
+                .andExpect(jsonPath("$.message", is("name is required")));
+
+        verifyNoInteractions(settingsService);
+    }
+
     @Test
     void deleteDataSourceShouldReturnSuccess() throws Exception {
         doNothing().when(settingsService).deleteDataSource("ds-1");
@@ -250,4 +285,20 @@ void testDataSourceShouldReturnTestResult() throws Exception {
                 .andExpect(jsonPath("$.data.success", is(true)))
                 .andExpect(jsonPath("$.data.message", is("Connection successful")));
     }
+
+    @Test
+    void testDataSourceShouldRejectMissingType() throws Exception {
+        mockMvc.perform(post("/api/settings/datasources/test")
+                        .contentType(MediaType.APPLICATION_JSON)
+                        .content("""
+                                {
+                                  "url": "localhost:9876"
+                                }
+                                """))
+                .andExpect(status().isBadRequest())
+                .andExpect(jsonPath("$.code", is(400)))
+                .andExpect(jsonPath("$.message", is("type is required")));
+
+        verifyNoInteractions(settingsService);
+    }
 }