feat(dlm): support filesystem

pom.xml

@@ -113,13 +113,13 @@
         <aliyun.ram-auth-client.version>1.2.19</aliyun.ram-auth-client.version>
         <jettison.version>1.5.4</jettison.version>
         <!-- ob-loader-dumper involved -->
-        <hadoop-common.version>3.3.4</hadoop-common.version>
+        <hadoop-common.version>3.3.6</hadoop-common.version>
         <netty.version>4.1.94.Final</netty.version>
         <obkv-table-client.version>1.2.1</obkv-table-client.version>
         <mina.version>2.1.6</mina.version>
 
         <!-- data-lifecycle-manager version -->
-        <data-lifecycle-manager.version>1.1.6.bp4</data-lifecycle-manager.version>
+        <data-lifecycle-manager.version>1.2.0</data-lifecycle-manager.version>
 
         <!-- plugin version -->
         <formatter-maven-plugin.version>2.11.0</formatter-maven-plugin.version>

server/odc-core/src/main/java/com/oceanbase/odc/core/shared/constant/ConnectType.java

@@ -35,6 +35,10 @@ public enum ConnectType {
     // reserved for future version
     ODP_SHARDING_OB_ORACLE(DialectType.OB_ORACLE),
     ORACLE(DialectType.ORACLE),
+    OSS(DialectType.FILE_SYSTEM),
+    OBS(DialectType.FILE_SYSTEM),
+    COS(DialectType.FILE_SYSTEM),
+    S3A(DialectType.FILE_SYSTEM),
     UNKNOWN(DialectType.UNKNOWN),
 
     ;

server/odc-core/src/main/java/com/oceanbase/odc/core/shared/constant/DialectType.java

@@ -29,6 +29,7 @@ public enum DialectType {
     ODP_SHARDING_OB_MYSQL,
     DORIS,
     POSTGRESQL,
+    FILE_SYSTEM,
     UNKNOWN,
     ;
 

server/odc-core/src/main/java/com/oceanbase/odc/core/shared/constant/ErrorCodes.java

@@ -170,6 +170,7 @@ public enum ErrorCodes implements ErrorCode {
     // Schedule
     AlterScheduleExists,
     InvalidCronExpression,
+    ScheduleIntervalTooShort,
     UpdateNotAllowed,
     PauseNotAllowed,
     DeleteNotAllowed,
@@ -315,7 +316,13 @@ public enum ErrorCodes implements ErrorCode {
      * workspace
      */
     WorkspaceDatabaseUserTypeMustBeAdmin,
-    ;
+    /**
+     * oss
+     */
+    BucketNotExist,
+    InvalidAccessKeyId,
+    SignatureDoesNotMatch,
+    UnsupportedSyncTableStructure;
 
     @Override
     public String code() {

server/odc-core/src/main/resources/i18n/ErrorMessages.properties

@@ -201,6 +201,11 @@ com.oceanbase.odc.ErrorCodes.DatabaseAccessDenied=Database access is denied beca
 com.oceanbase.odc.ErrorCodes.ObQueryProfileNotSupported=Query profile is only available for OceanBase Database with versions equal to or higher than {0}.
 com.oceanbase.odc.ErrorCodes.WorksheetEditVersionConflict=Someone has just modified this file. Please refresh the page to get the latest version and continue editing.
 com.oceanbase.odc.ErrorCodes.WorkspaceDatabaseUserTypeMustBeAdmin=The database user type used in the workspace must be a super account.
+com.oceanbase.odc.ErrorCodes.BucketNotExist=Bucket does not exist
+com.oceanbase.odc.ErrorCodes.InvalidAccessKeyId=Invalid access key id
+com.oceanbase.odc.ErrorCodes.SignatureDoesNotMatch=Invalid access key secret
+com.oceanbase.odc.ErrorCodes.UnsupportedSyncTableStructure=Sync table structure is not supported for {0} to {1}
+com.oceanbase.odc.ErrorCodes.ScheduleIntervalTooShort=The execution interval is configured too short, please reconfigure. The minimum interval is: {0} seconds.
 com.oceanbase.odc.ErrorCodes.UpdateNotAllowed=Editing is not allowed in the current state. Please try again after disabling.
 com.oceanbase.odc.ErrorCodes.PauseNotAllowed=Disabling is not allowed in the current state, please check if there are any records in execution.
 com.oceanbase.odc.ErrorCodes.DeleteNotAllowed=Deletion is not allowed in the current state.

server/odc-core/src/main/resources/i18n/ErrorMessages_zh_CN.properties

@@ -204,3 +204,9 @@ com.oceanbase.odc.ErrorCodes.UpdateNotAllowed=当前状态下不允许编辑，
 com.oceanbase.odc.ErrorCodes.PauseNotAllowed=当前状态下不允许禁用，请检查是否存在执行中的记录
 com.oceanbase.odc.ErrorCodes.DeleteNotAllowed=当前状态下不允许删除
 
+
+com.oceanbase.odc.ErrorCodes.BucketNotExist=桶不存在
+com.oceanbase.odc.ErrorCodes.InvalidAccessKeyId=无效的 AccessKeyId
+com.oceanbase.odc.ErrorCodes.SignatureDoesNotMatch=无效的 AccessKeySecret
+com.oceanbase.odc.ErrorCodes.UnsupportedSyncTableStructure=结构同步暂不支持 {0} 到 {1}
+com.oceanbase.odc.ErrorCodes.ScheduleIntervalTooShort=执行间隔配置过短，请重新配置。最小间隔为：{0} 秒

server/odc-core/src/main/resources/i18n/ErrorMessages_zh_TW.properties

@@ -205,3 +205,6 @@ com.oceanbase.odc.ErrorCodes.PauseNotAllowed=當前狀態下不允許禁用，
 com.oceanbase.odc.ErrorCodes.DeleteNotAllowed=在目前狀態下不允許删除
 
 
+
+com.oceanbase.odc.ErrorCodes.UnsupportedSyncTableStructure=結構同步暫不支持 {0} 到 {1}
+com.oceanbase.odc.ErrorCodes.ScheduleIntervalTooShort=執行間隔設定過短，請重新設定。最小間隔應為：{0} 秒

server/odc-server/src/main/resources/data.sql

@@ -859,4 +859,10 @@ INSERT INTO config_system_configuration(`key`, `value`, `description`) VALUES('o
 INSERT INTO `config_system_configuration` (`key`, `value`, `application`, `profile`, `label`, `description`)
 VALUES ('odc.session.kill-query-or-session.max-supported-ob-version', '4.2.5', 'odc', 'default', 'master',
         'Max OBVersion kill session or kill query supported, only take effect when value greater than 0')
-ON DUPLICATE KEY UPDATE `id`=`id`;
+ON DUPLICATE KEY UPDATE `id`=`id`;
+INSERT INTO config_system_configuration ( `key`, `value`, `description` ) VALUES('odc.task.dlm.session-limiting.enabled', 'true',
+'Explosion-proof current limiting switch of mysql/oracle' )
+ON DUPLICATE KEY UPDATE `id` = `id`;
+INSERT INTO config_system_configuration ( `key`, `value`, `description` ) VALUES('odc.task.dlm.session-limiting-ratio', '25',
+'The ratio of oracle/mysql active sessions to the maximum number of connections allowed' )
+ON DUPLICATE KEY UPDATE `id` = `id`;

server/odc-server/src/main/resources/log4j2-task.xml

@@ -64,6 +64,11 @@
                             <!-- TRACE < DEBUG < INFO < WARN < ERROR < FATAL -->
                             <ThresholdFilter level="INFO" onMatch="ACCEPT" onMismatch="DENY"/>
                         </Filters>
+                        <Policies>
+                            <OnStartupTriggeringPolicy/>
+                            <TimeBasedTriggeringPolicy interval="6" modulate="true"/>
+                            <SizeBasedTriggeringPolicy size="10 MB"/>
+                        </Policies>
                     </RollingFile>
                 </Route>
             </Routes>
@@ -87,6 +92,11 @@
                             <!-- TRACE < DEBUG < INFO < WARN < ERROR < FATAL -->
                             <ThresholdFilter level="WARN" onMatch="ACCEPT" onMismatch="DENY"/>
                         </Filters>
+                        <Policies>
+                            <OnStartupTriggeringPolicy/>
+                            <TimeBasedTriggeringPolicy interval="6" modulate="true"/>
+                            <SizeBasedTriggeringPolicy size="10 MB"/>
+                        </Policies>
                     </RollingFile>
                 </Route>
             </Routes>
@@ -117,6 +127,9 @@
         <Logger name="com.oceanbase.odc.service.plugin.PluginProperties" level="INFO" additivity="false">
             <AppenderRef ref="ExecutorFileAppender"/>
         </Logger>
+        <Logger name="com.oceanbase.odc.agent" level="INFO" additivity="false">
+            <AppenderRef ref="ExecutorFileAppender"/>
+        </Logger>
         <!-- 执行器日志输出 end -->
 
         <!-- 任务日志输出，输出级别 INFO -->

server/odc-service/src/main/java/com/oceanbase/odc/service/connection/ConnectionEventPublisher.java

@@ -0,0 +1,52 @@
+/*
+ * Copyright (c) 2023 OceanBase.
+ *
+ * Licensed 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 com.oceanbase.odc.service.connection;
+
+import javax.annotation.PostConstruct;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Component;
+
+import com.oceanbase.odc.common.event.AbstractEvent;
+import com.oceanbase.odc.common.event.LocalEventPublisher;
+import com.oceanbase.odc.service.connection.listener.UpdateDatasourceListener;
+
+import lombok.NonNull;
+
+/**
+ * @Author：tinker
+ * @Date: 2024/12/30 10:57
+ * @Descripition:
+ */
+
+@Component
+public class ConnectionEventPublisher {
+    @Autowired
+    private LocalEventPublisher localEventPublisher;
+
+    @Autowired
+    private UpdateDatasourceListener updateDatasourceListener;
+
+    @PostConstruct
+    public void init() {
+        localEventPublisher.addEventListener(updateDatasourceListener);
+    }
+
+    public <T extends AbstractEvent> void publishEvent(@NonNull T event) {
+        localEventPublisher.publishEvent(event);
+    }
+
+}

server/odc-service/src/main/java/com/oceanbase/odc/service/connection/ConnectionService.java

@@ -105,6 +105,7 @@
 import com.oceanbase.odc.service.connection.ConnectionStatusManager.CheckState;
 import com.oceanbase.odc.service.connection.database.DatabaseService;
 import com.oceanbase.odc.service.connection.database.DatabaseSyncManager;
+import com.oceanbase.odc.service.connection.event.UpsertDatasourceEvent;
 import com.oceanbase.odc.service.connection.model.ConnectProperties;
 import com.oceanbase.odc.service.connection.model.ConnectionConfig;
 import com.oceanbase.odc.service.connection.model.OBTenantEndpoint;
@@ -221,6 +222,9 @@ public class ConnectionService {
     @Autowired
     private TransactionTemplate txTemplate;
 
+    @Autowired
+    private ConnectionEventPublisher connectionEventPublisher;
+
     private final ConnectionMapper mapper = ConnectionMapper.INSTANCE;
 
     public static final String DEFAULT_MIN_PRIVILEGE = "read";
@@ -249,6 +253,7 @@ public ConnectionConfig create(@NotNull @Valid ConnectionConfig connection, @Not
             }
         });
         databaseSyncManager.submitSyncDataSourceAndDBSchemaTask(saved);
+        connectionEventPublisher.publishEvent(new UpsertDatasourceEvent(saved));
         return saved;
     }
 
@@ -393,8 +398,11 @@ public ConnectionConfig getWithoutPermissionCheck(@NotNull Long id) {
 
     @SkipAuthorize("odc internal usage")
     public List<ConnectionConfig> listByOrganizationId(@NonNull Long organizationId) {
-        return entitiesToModels(repository.findByOrganizationIdOrderByNameAsc(organizationId), organizationId, true,
+        List<ConnectionConfig> connectionConfigs = entitiesToModels(
+                repository.findByOrganizationIdOrderByNameAsc(organizationId), organizationId, true,
                 true);
+        fullFillAttributes(connectionConfigs);
+        return connectionConfigs;
     }
 
     @SkipAuthorize("odc internal usage")
@@ -685,6 +693,7 @@ private ConnectionConfig updateConnectionConfig(Long id, ConnectionConfig connec
             }
         });
         databaseSyncManager.submitSyncDataSourceAndDBSchemaTask(config);
+        connectionEventPublisher.publishEvent(new UpsertDatasourceEvent(config));
         return config;
     }
 

server/odc-service/src/main/java/com/oceanbase/odc/service/connection/ConnectionTesting.java

@@ -74,6 +74,8 @@ public class ConnectionTesting {
     private ConnectionSSLAdaptor connectionSSLAdaptor;
     @Autowired
     private CloudMetadataClient cloudMetadataClient;
+    @Autowired
+    private FileSystemConnectionTester fileSystemConnectionTesting;
     @Value("${odc.sdk.test-connect.query-timeout-seconds:2}")
     private int queryTimeoutSeconds = 2;
 
@@ -102,6 +104,9 @@ public ConnectionTestResult test(@NotNull @Valid TestConnectionReq req) {
 
     public ConnectionTestResult test(@NonNull ConnectionConfig config) {
         ConnectType type = config.getType();
+        if (type.getDialectType() == DialectType.FILE_SYSTEM) {
+            return fileSystemConnectionTesting.test(config);
+        }
         try {
             /**
              * 进行连接测试时需要关注的值有一个 {@link ConnectType}， 容易产生问题信息主要是两个：{@code username}, {@code defaultSchema} 首先分析
@@ -232,6 +237,7 @@ private ConnectionConfig reqToConnectionConfig(TestConnectionReq req) {
         config.setServiceName(req.getServiceName());
         config.setUserRole(req.getUserRole());
         config.setCatalogName(req.getCatalogName());
+        config.setRegion(req.getRegion());
 
         OBTenantEndpoint endpoint = req.getEndpoint();
         if (Objects.nonNull(endpoint) && OceanBaseAccessMode.IC_PROXY == endpoint.getAccessMode()) {

server/odc-service/src/main/java/com/oceanbase/odc/service/connection/ConnectionValidator.java

@@ -21,6 +21,7 @@
 import org.springframework.stereotype.Component;
 
 import com.oceanbase.odc.core.shared.PreConditions;
+import com.oceanbase.odc.core.shared.constant.DialectType;
 import com.oceanbase.odc.core.shared.constant.ErrorCodes;
 import com.oceanbase.odc.core.shared.exception.AccessDeniedException;
 import com.oceanbase.odc.service.collaboration.environment.EnvironmentService;
@@ -43,7 +44,9 @@ public class ConnectionValidator {
     void validateForUpsert(ConnectionConfig connection) {
         PreConditions.notNull(connection, "connection");
         PreConditions.notBlank(connection.getHost(), "connection.host");
-        PreConditions.notNull(connection.getPort(), "connection.port");
+        if (connection.getDialectType() != DialectType.FILE_SYSTEM) {
+            PreConditions.notNull(connection.getPort(), "connection.port");
+        }
         PreConditions.validNotSqlInjection(connection.getUsername(), "username");
         PreConditions.validNotSqlInjection(connection.getClusterName(), "clusterName");
         PreConditions.validNotSqlInjection(connection.getTenantName(), "tenantName");