RockteMQ-AI commented on code in PR #4352:
URL:
https://github.com/apache/rocketmq-dashboard/pull/4352#discussion_r4017462952
##########
server/src/main/java/org/apache/rocketmq/studio/ops/OpsService.java:
##########
@@ -30,49 +33,165 @@ public class OpsService {
private static final String OPS_SETTINGS_UNAVAILABLE =
"Ops settings are not connected to the cluster admin
configuration";
+ private final OpsRuntimeConnection runtimeConnection;
+ private final OpsRuntimeProperties runtimeProperties;
+ private final OperationAuditService auditService;
+
+ public OpsService(OpsRuntimeConnection runtimeConnection,
OpsRuntimeProperties runtimeProperties,
+ OperationAuditService auditService) {
+ this.runtimeConnection = runtimeConnection;
+ this.runtimeProperties = runtimeProperties;
+ this.auditService = auditService;
+ }
Review Comment:
Proper fail-closed design: when runtimeProperties.isEnabled() is false, the
service returns unavailable status instead of silently falling back to
potentially stale config.
##########
server/src/main/java/org/apache/rocketmq/studio/ops/MybatisPlusOpsConnectionRepository.java:
##########
@@ -0,0 +1,172 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.studio.ops;
+
+import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.rocketmq.studio.common.exception.BusinessException;
+import org.apache.rocketmq.studio.persistence.entity.RmqSettings;
+import org.apache.rocketmq.studio.persistence.mapper.RmqSettingsMapper;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.dao.DuplicateKeyException;
+import org.springframework.stereotype.Repository;
+import org.springframework.transaction.annotation.Transactional;
+
+import java.time.LocalDateTime;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.function.Consumer;
+import java.util.function.Supplier;
+import java.util.function.UnaryOperator;
+
+@Slf4j
+@Repository
+public class MybatisPlusOpsConnectionRepository implements
OpsConnectionRepository {
+
+ static final String OPS_CONNECTION_SETTINGS_KEY = "ops-connection";
+
+ private final RmqSettingsMapper settingsMapper;
+ private final ObjectMapper objectMapper;
+
+ public MybatisPlusOpsConnectionRepository(RmqSettingsMapper
settingsMapper, ObjectMapper objectMapper) {
+ this.settingsMapper = settingsMapper;
+ this.objectMapper = objectMapper;
+ }
+
+ @Override
+ public Optional<OpsConnectionSettings> load() {
+ RmqSettings entity = findSettings();
+ if (entity == null || entity.getJson() == null) {
+ return Optional.empty();
+ }
+ try {
+ OpsConnectionSettings settings =
objectMapper.readValue(entity.getJson(), OpsConnectionSettings.class);
+ if (settings == null) {
+ throw new BusinessException(500, "Persisted Ops runtime
settings are invalid");
+ }
+ return Optional.of(settings);
+ } catch (JsonProcessingException | IllegalArgumentException exception)
{
+ log.error("Failed to deserialize Ops runtime settings", exception);
+ throw new BusinessException(500, "Persisted Ops runtime settings
are invalid");
+ }
+ }
+
+ @Override
+ @Transactional
+ public void save(OpsConnectionSettings settings) {
+ String json = toJson(settings);
+ RmqSettings entity = findSettings();
+ LocalDateTime now = LocalDateTime.now();
+ if (entity == null) {
+ try {
+ insertRaw(json, now);
+ } catch (DuplicateKeyException duplicateKey) {
+ RmqSettings concurrent = findSettings();
+ if (concurrent == null) {
+ throw duplicateKey;
+ }
+ concurrent.setJson(json);
+ concurrent.setGmtModified(LocalDateTime.now());
+ settingsMapper.updateById(concurrent);
+ }
+ } else {
+ entity.setJson(json);
+ entity.setGmtModified(now);
+ settingsMapper.updateById(entity);
+ }
+ }
+
+ @Override
+ @Transactional
+ public OpsConnectionSettings update(UnaryOperator<OpsConnectionSettings>
updater,
+ Supplier<OpsConnectionSettings>
defaultSettings,
+ Consumer<OpsConnectionSettings>
validator) {
+ return updateLocked(updater, defaultSettings, validator, false);
+ }
+
+ private RmqSettings findSettings() {
+ return settingsMapper.selectOne(new QueryWrapper<RmqSettings>()
+ .eq("settings_key", OPS_CONNECTION_SETTINGS_KEY)
+ .last("LIMIT 1"));
+ }
+
+ private RmqSettings findSettingsForUpdate() {
+ return settingsMapper.selectOne(new QueryWrapper<RmqSettings>()
+ .eq("settings_key", OPS_CONNECTION_SETTINGS_KEY)
+ .last("LIMIT 1 FOR UPDATE"));
+ }
+
+ private OpsConnectionSettings
updateLocked(UnaryOperator<OpsConnectionSettings> updater,
+ Supplier<OpsConnectionSettings>
defaultSettings,
+ Consumer<OpsConnectionSettings>
validator,
+ boolean
retriedAfterConcurrentInsert) {
Review Comment:
Good use of SELECT ... FOR UPDATE with retry logic for concurrent inserts.
The retriedAfterConcurrentInsert flag prevents infinite loops on persistent
conflicts.
##########
server/src/main/java/org/apache/rocketmq/studio/ops/OpsService.java:
##########
@@ -30,49 +33,165 @@ public class OpsService {
private static final String OPS_SETTINGS_UNAVAILABLE =
"Ops settings are not connected to the cluster admin
configuration";
+ private final OpsRuntimeConnection runtimeConnection;
+ private final OpsRuntimeProperties runtimeProperties;
+ private final OperationAuditService auditService;
+
+ public OpsService(OpsRuntimeConnection runtimeConnection,
OpsRuntimeProperties runtimeProperties,
+ OperationAuditService auditService) {
+ this.runtimeConnection = runtimeConnection;
+ this.runtimeProperties = runtimeProperties;
+ this.auditService = auditService;
+ }
+
public synchronized OpsHomeVO getHomePage() {
- return OpsHomeVO.builder()
- .configurationAvailable(false)
- .unavailableReason(OPS_SETTINGS_UNAVAILABLE)
- .namesvrAddrList(List.of())
- .currentNamesrv("")
- .useVIPChannel(false)
- .useTLS(false)
- .build();
+ if (!runtimeProperties.isEnabled()) {
+ return unavailableHomePage(OPS_SETTINGS_UNAVAILABLE);
+ }
+ try {
+ OpsConnectionSettings settings = runtimeConnection.current();
+ if (settings.currentNamesrv().isEmpty()) {
+ return unavailableHomePage("namesrvAddr is required");
+ }
+ return OpsHomeVO.builder()
+ .configurationAvailable(true)
+ .namesvrAddrList(settings.addresses())
+ .currentNamesrv(settings.currentNamesrv())
+ .useVIPChannel(settings.useVIPChannel())
+ .useTLS(settings.useTLS())
+ .build();
+ } catch (BusinessException exception) {
+ return unavailableHomePage(exception.getMessage());
+ }
}
Review Comment:
Good security practice: ensureAdmin() is called before any write operation,
preventing unauthorized changes even when global login is disabled.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
