diff --git a/configure/etc/conf/plugins.properties b/configure/etc/conf/plugins.properties index 73e3fc39..6603804c 100644 --- a/configure/etc/conf/plugins.properties +++ b/configure/etc/conf/plugins.properties @@ -1,5 +1,6 @@ datacap-plugin-mysql=plugin/datacap-plugin-mysql/pom.xml datacap-plugin-clickhouse=plugin/datacap-plugin-clickhouse/pom.xml +datacap-plugin-influxdb=plugin/datacap-plugin-influxdb/pom.xml datacap-convert-txt=convert/datacap-convert-txt/pom.xml datacap-convert-json=convert/datacap-convert-json/pom.xml datacap-convert-xml=convert/datacap-convert-xml/pom.xml diff --git a/configure/metadata.json b/configure/metadata.json index e292f310..1cb3913d 100644 --- a/configure/metadata.json +++ b/configure/metadata.json @@ -35,6 +35,21 @@ ], "url": "https://cdn.north.devlive.org/applications/datacap/plugins/2024.4.0-SNAPSHOT/plugin/datacap-plugin-clickhouse-bin.tar.gz" }, + { + "key": "datacap-plugin-influxdb", + "label": "InfluxDB", + "description": "InfluxDB is a time series database that stores and retrieves data points. It is a distributed database with a focus on performance, scalability, and ease of use.", + "i18nFormat": true, + "type": "Connector", + "version": "2024.4.0-SNAPSHOT", + "author": "datacap-community", + "logo": "https://cdn.north.devlive.org/applications/datacap/resources/logo/plugin/influxdb.svg", + "released": "2024-11-21 23:11:15", + "supportVersion": [ + "ALL" + ], + "url": "https://cdn.north.devlive.org/applications/datacap/plugins/2024.4.0-SNAPSHOT/plugin/datacap-plugin-influxdb-bin.tar.gz" + }, { "key": "datacap-convert-csv", "label": "CsvConvert", diff --git a/core/datacap-plugin/src/main/java/io/edurt/datacap/plugin/loader/PropertiesPluginLoader.java b/core/datacap-plugin/src/main/java/io/edurt/datacap/plugin/loader/PropertiesPluginLoader.java index 81a30de7..01ea9c41 100644 --- a/core/datacap-plugin/src/main/java/io/edurt/datacap/plugin/loader/PropertiesPluginLoader.java +++ b/core/datacap-plugin/src/main/java/io/edurt/datacap/plugin/loader/PropertiesPluginLoader.java @@ -7,6 +7,7 @@ import io.edurt.datacap.plugin.utils.PluginPathUtils; import lombok.extern.slf4j.Slf4j; import java.io.FileInputStream; +import java.io.IOException; import java.nio.file.Path; import java.nio.file.Paths; import java.util.ArrayList; @@ -81,7 +82,7 @@ public class PropertiesPluginLoader } } } - catch (Exception e) { + catch (IOException e) { log.error("Failed to load plugins from properties file: {}", path, e); } return plugins; diff --git a/core/datacap-plugin/src/main/java/io/edurt/datacap/plugin/loader/TarPluginLoader.java b/core/datacap-plugin/src/main/java/io/edurt/datacap/plugin/loader/TarPluginLoader.java index f73ca3d0..b4392892 100644 --- a/core/datacap-plugin/src/main/java/io/edurt/datacap/plugin/loader/TarPluginLoader.java +++ b/core/datacap-plugin/src/main/java/io/edurt/datacap/plugin/loader/TarPluginLoader.java @@ -93,31 +93,37 @@ public class TarPluginLoader public List load(Path path, Path targetDir, Set parentClassLoaderPackages) { try { - // 如果是 URL 路径，先下载到本地 - // If it's a URL path, download it first - if (path.toString().startsWith("http") || path.toString().startsWith("https")) { - path = downloadTarFile(path.toString().replace(":/", "://")); + if (isValidTarPath(path)) { + // 如果是 URL 路径，先下载到本地 + // If it's a URL path, download it first + if (path.toString().startsWith("http") || path.toString().startsWith("https")) { + path = downloadTarFile(path.toString().replace(":/", "://")); + } + + if (isValidTarFile(path)) { + // 使用指定的目录或创建临时目录 + // Use specified directory or create temporary directory + Path extractDir = targetDir != null ? targetDir : createTempDirectory(); + + // 解压 tar 文件 + // Extract tar file + extractTarFile(path, extractDir); + + // 从解压目录加载插件 + // Load plugins from extracted directory + List plugins = loadPluginsFromDirectory(extractDir, parentClassLoaderPackages); + + // 如果使用的是临时目录，则清理 + // Clean up if using temporary directory + if (targetDir == null) { + cleanupTempDirectory(extractDir); + } + + return plugins; + } } - // 使用指定的目录或创建临时目录 - // Use specified directory or create temporary directory - Path extractDir = targetDir != null ? targetDir : createTempDirectory(); - - // 解压 tar 文件 - // Extract tar file - extractTarFile(path, extractDir); - - // 从解压目录加载插件 - // Load plugins from extracted directory - List plugins = loadPluginsFromDirectory(extractDir, parentClassLoaderPackages); - - // 如果使用的是临时目录，则清理 - // Clean up if using temporary directory - if (targetDir == null) { - cleanupTempDirectory(extractDir); - } - - return plugins; + return List.of(); } catch (Exception e) { log.error("Failed to load plugins from tar file: {}", path, e); @@ -327,4 +333,39 @@ public class TarPluginLoader log.warn("Failed to cleanup temporary directory: {}", directory, e); } } + + /** + * 检查是否为合法 Tar 文件路径 + * Check if it's a valid Tar file path + * + * @param path file path + * @return true if it's a valid Tar file + */ + private boolean isValidTarPath(Path path) + { + String pathStr = path.toString(); + return (pathStr.startsWith("http") || pathStr.startsWith("https")) + && (pathStr.endsWith("tar.gz") || pathStr.endsWith("tar") || pathStr.endsWith("tgz")); + } + + /** + * 检查是否合法的 Tar 文件，通过解压文件头检查 + * Check if it's a valid Tar file, by checking the file header + * + * @param path file path + * @return true if it's a valid Tar file + */ + private boolean isValidTarFile(Path path) + { + try (InputStream fileIn = Files.newInputStream(path); + BufferedInputStream buffIn = new BufferedInputStream(fileIn); + GZIPInputStream gzipIn = new GZIPInputStream(buffIn)) { + log.debug("Validating tar file: {}", path); + log.debug("Tar file header: {}", gzipIn.readNBytes(512)); + return true; + } + catch (IOException e) { + return false; + } + } } diff --git a/core/datacap-server/pom.xml b/core/datacap-server/pom.xml index c4b8f66b..828c19f2 100644 --- a/core/datacap-server/pom.xml +++ b/core/datacap-server/pom.xml @@ -421,12 +421,6 @@ - - - - - - diff --git a/core/datacap-spi/src/main/java/io/edurt/datacap/spi/PluginService.java b/core/datacap-spi/src/main/java/io/edurt/datacap/spi/PluginService.java index dd5de03a..11349ef6 100644 --- a/core/datacap-spi/src/main/java/io/edurt/datacap/spi/PluginService.java +++ b/core/datacap-spi/src/main/java/io/edurt/datacap/spi/PluginService.java @@ -6,14 +6,17 @@ import io.edurt.datacap.spi.adapter.HttpAdapter; import io.edurt.datacap.spi.adapter.JdbcAdapter; import io.edurt.datacap.spi.adapter.NativeAdapter; import io.edurt.datacap.spi.connection.Connection; -import io.edurt.datacap.spi.connection.JdbcConfigure; import io.edurt.datacap.spi.connection.JdbcConnection; import io.edurt.datacap.spi.model.Configure; import io.edurt.datacap.spi.model.Response; -import org.apache.commons.beanutils.BeanUtils; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.stream.Collectors; + public interface PluginService extends Service { @@ -54,23 +57,64 @@ public interface PluginService { Response response = new Response(); try { - JdbcConfigure jdbcConfigure = new JdbcConfigure(); - if (jdbcConfigure.getJdbcDriver() == null) { - jdbcConfigure.setJdbcDriver(this.driver()); - } - if (jdbcConfigure.getJdbcType() == null) { - jdbcConfigure.setJdbcType(this.connectType()); - } - BeanUtils.copyProperties(jdbcConfigure, configure); - local.set(new JdbcConnection(jdbcConfigure, response)); + configure.setDriver(this.driver()); + configure.setType(this.connectType()); + configure.setUrl(Optional.of(url(configure))); + local.set(new JdbcConnection(configure, response)); } catch (Exception ex) { response.setIsConnected(Boolean.FALSE); response.setMessage(ex.getMessage()); - log.error("Error connecting : {}", ex); + log.error("Error connecting :", ex); } } + /** + * 构建驱动路径 + * Build the driver path + * + * @param configure 配置信息 | Configuration information + * @return 驱动路径 | Driver path + */ + default String url(Configure configure) + { + if (configure.getUrl().isPresent()) { + return configure.getUrl().get(); + } + + StringBuilder buffer = new StringBuilder(); + buffer.append("jdbc:"); + buffer.append(configure.getType()); + buffer.append("://"); + buffer.append(configure.getHost()); + buffer.append(":"); + buffer.append(configure.getPort()); + if (configure.getDatabase().isPresent()) { + buffer.append("/"); + buffer.append(configure.getDatabase().get()); + } + if (configure.getSsl().isPresent()) { + buffer.append(String.format("?ssl=%s", configure.getSsl().get())); + } + if (configure.getEnv().isPresent()) { + Map env = configure.getEnv().get(); + List flatEnv = env.entrySet() + .stream() + .map(value -> String.format("%s=%s", value.getKey(), value.getValue())) + .collect(Collectors.toList()); + if (configure.getSsl().isEmpty()) { + buffer.append("?"); + } + else { + if (configure.getIsAppendChar()) { + buffer.append("&"); + } + } + buffer.append(String.join("&", flatEnv)); + } + return buffer.toString(); + } + @Deprecated default Response execute(String content) { diff --git a/core/datacap-spi/src/main/java/io/edurt/datacap/spi/adapter/JdbcAdapter.java b/core/datacap-spi/src/main/java/io/edurt/datacap/spi/adapter/JdbcAdapter.java index ec1e6444..84c9b7fb 100644 --- a/core/datacap-spi/src/main/java/io/edurt/datacap/spi/adapter/JdbcAdapter.java +++ b/core/datacap-spi/src/main/java/io/edurt/datacap/spi/adapter/JdbcAdapter.java @@ -3,8 +3,8 @@ package io.edurt.datacap.spi.adapter; import edu.umd.cs.findbugs.annotations.SuppressFBWarnings; import io.edurt.datacap.spi.column.Column; import io.edurt.datacap.spi.column.JdbcColumn; -import io.edurt.datacap.spi.connection.JdbcConfigure; import io.edurt.datacap.spi.connection.JdbcConnection; +import io.edurt.datacap.spi.model.Configure; import io.edurt.datacap.spi.model.Response; import io.edurt.datacap.spi.model.Time; import lombok.extern.slf4j.Slf4j; @@ -39,7 +39,7 @@ public class JdbcAdapter processorTime.setStart(new Date().getTime()); Response response = jdbcConnection.getResponse(); Connection connection = (Connection) jdbcConnection.getConnection(); - JdbcConfigure configure = (JdbcConfigure) jdbcConnection.getConfigure(); + Configure configure = jdbcConnection.getConfigure(); if (response.getIsConnected()) { try (Statement statement = connection.createStatement()) { List headers = new ArrayList<>(); diff --git a/core/datacap-spi/src/main/java/io/edurt/datacap/spi/connection/Connection.java b/core/datacap-spi/src/main/java/io/edurt/datacap/spi/connection/Connection.java index 2f8a0f95..50aa5180 100644 --- a/core/datacap-spi/src/main/java/io/edurt/datacap/spi/connection/Connection.java +++ b/core/datacap-spi/src/main/java/io/edurt/datacap/spi/connection/Connection.java @@ -25,8 +25,6 @@ public abstract class Connection this.response.setConnection(connectionTime); } - protected abstract String formatJdbcUrl(); - protected abstract java.sql.Connection openConnection(); public Object getConnection() diff --git a/core/datacap-spi/src/main/java/io/edurt/datacap/spi/connection/JdbcConfigure.java b/core/datacap-spi/src/main/java/io/edurt/datacap/spi/connection/JdbcConfigure.java deleted file mode 100644 index cd16584e..00000000 --- a/core/datacap-spi/src/main/java/io/edurt/datacap/spi/connection/JdbcConfigure.java +++ /dev/null @@ -1,20 +0,0 @@ -package io.edurt.datacap.spi.connection; - -import edu.umd.cs.findbugs.annotations.SuppressFBWarnings; -import io.edurt.datacap.spi.model.Configure; -import lombok.Data; -import lombok.EqualsAndHashCode; -import lombok.ToString; - -@Data -@ToString -@EqualsAndHashCode -@SuppressFBWarnings(value = {"EQ_OVERRIDING_EQUALS_NOT_SYMMETRIC"}, - justification = "I prefer to suppress these FindBugs warnings") -public class JdbcConfigure - extends Configure -{ - private String jdbcDriver; - private String jdbcType; - private Boolean isAppendChar = Boolean.TRUE; -} diff --git a/core/datacap-spi/src/main/java/io/edurt/datacap/spi/connection/JdbcConnection.java b/core/datacap-spi/src/main/java/io/edurt/datacap/spi/connection/JdbcConnection.java index a4509c4d..020970c1 100644 --- a/core/datacap-spi/src/main/java/io/edurt/datacap/spi/connection/JdbcConnection.java +++ b/core/datacap-spi/src/main/java/io/edurt/datacap/spi/connection/JdbcConnection.java @@ -2,6 +2,7 @@ package io.edurt.datacap.spi.connection; import io.edurt.datacap.plugin.PluginContextManager; import io.edurt.datacap.plugin.loader.PluginClassLoader; +import io.edurt.datacap.spi.model.Configure; import io.edurt.datacap.spi.model.Response; import lombok.extern.slf4j.Slf4j; import org.apache.commons.lang3.ObjectUtils; @@ -11,11 +12,8 @@ import java.sql.Driver; import java.sql.DriverManager; import java.sql.DriverPropertyInfo; import java.sql.SQLException; -import java.util.List; -import java.util.Map; import java.util.Properties; import java.util.logging.Logger; -import java.util.stream.Collectors; @Slf4j public class JdbcConnection @@ -23,88 +21,41 @@ public class JdbcConnection { private java.sql.Connection jdbcConnection; - public JdbcConnection(JdbcConfigure configure, Response response) + public JdbcConnection(Configure configure, Response response) { super(configure, response); } - protected String formatJdbcUrl() - { - JdbcConfigure jdbcConfigure = getJdbcConfigure(); - StringBuilder buffer = new StringBuilder(); - buffer.append("jdbc:"); - buffer.append(jdbcConfigure.getJdbcType()); - if (jdbcConfigure.getJdbcType().equals("influxdb")) { - buffer.append(":"); - } - else { - buffer.append("://"); - } - buffer.append(jdbcConfigure.getHost()); - buffer.append(":"); - buffer.append(jdbcConfigure.getPort()); - if (jdbcConfigure.getDatabase().isPresent()) { - if (jdbcConfigure.getJdbcType().equals("solr")) { - buffer.append("/?collection="); - } - else { - buffer.append("/"); - } - buffer.append(jdbcConfigure.getDatabase().get()); - } - if (jdbcConfigure.getSsl().isPresent()) { - buffer.append(String.format("?ssl=%s", jdbcConfigure.getSsl().get())); - } - if (jdbcConfigure.getEnv().isPresent()) { - Map env = jdbcConfigure.getEnv().get(); - List flatEnv = env.entrySet() - .stream() - .map(value -> String.format("%s=%s", value.getKey(), value.getValue())) - .collect(Collectors.toList()); - if (jdbcConfigure.getSsl().isEmpty()) { - buffer.append("?"); - } - else { - if (jdbcConfigure.getIsAppendChar()) { - buffer.append("&"); - } - } - buffer.append(String.join("&", flatEnv)); - } - return buffer.toString(); - } - - protected JdbcConfigure getJdbcConfigure() - { - return (JdbcConfigure) this.configure; - } - protected java.sql.Connection openConnection() { - JdbcConfigure jdbcConfigure = getJdbcConfigure(); - try { - PluginClassLoader pluginClassLoader = jdbcConfigure.getPlugin().getPluginClassLoader(); + PluginClassLoader pluginClassLoader = configure.getPlugin().getPluginClassLoader(); PluginContextManager.runWithClassLoader(pluginClassLoader, () -> { - Class driverClass = Class.forName(jdbcConfigure.getJdbcDriver(), true, pluginClassLoader); + Class driverClass = Class.forName(configure.getDriver(), true, pluginClassLoader); Driver driver = (Driver) driverClass.getDeclaredConstructor().newInstance(); DriverManager.registerDriver(new DriverShim(driver)); - String url = formatJdbcUrl(); - log.info("Connection driver {}", jdbcConfigure.getJdbcDriver()); + if (configure.getUrl().isEmpty()) { + throw new RuntimeException("Connection url not present"); + } + + String url = configure.getUrl().get(); + log.info("Connection driver {}", configure.getType()); log.info("Connection url {}", url); - if (jdbcConfigure.getUsername().isPresent() && jdbcConfigure.getPassword().isPresent()) { + if (configure.getUsername().isPresent() && configure.getPassword().isPresent()) { log.info("Connection username with {} password with {}", - jdbcConfigure.getUsername().get(), "***"); - this.jdbcConnection = DriverManager.getConnection(url, - jdbcConfigure.getUsername().get(), - jdbcConfigure.getPassword().get()); + configure.getUsername().get(), "***"); + this.jdbcConnection = DriverManager.getConnection( + url, + configure.getUsername().get(), + configure.getPassword().get() + ); } else { log.info("Connection username and password not present"); Properties properties = new Properties(); - if (jdbcConfigure.getUsername().isPresent()) { - properties.put("user", jdbcConfigure.getUsername().get()); + if (configure.getUsername().isPresent()) { + properties.put("user", configure.getUsername().get()); } this.jdbcConnection = DriverManager.getConnection(url, properties); } diff --git a/core/datacap-spi/src/main/java/io/edurt/datacap/spi/model/Configure.java b/core/datacap-spi/src/main/java/io/edurt/datacap/spi/model/Configure.java index eddcfa6b..d59088ad 100644 --- a/core/datacap-spi/src/main/java/io/edurt/datacap/spi/model/Configure.java +++ b/core/datacap-spi/src/main/java/io/edurt/datacap/spi/model/Configure.java @@ -27,12 +27,26 @@ public class Configure private PluginManager pluginManager; private String host; private Integer port; + + @Builder.Default private Optional username = Optional.empty(); + + @Builder.Default private Optional password = Optional.empty(); + + @Builder.Default private Optional database = Optional.empty(); + + @Builder.Default private Optional version = Optional.empty(); + + @Builder.Default private Optional> env = Optional.empty(); + + @Builder.Default private Optional ssl = Optional.empty(); + + @Builder.Default private String format = "JsonConvert"; // if `to`: skip private Optional query = Optional.empty(); @@ -40,4 +54,15 @@ public class Configure private String home; private boolean usedConfig; private String id; + + // 自定义 url + // Custom url + @Builder.Default + private Optional url = Optional.empty(); + + private String driver; + private String type; + + @Builder.Default + private Boolean isAppendChar = Boolean.TRUE; } diff --git a/core/datacap-ui/public/static/images/plugin/ClickHouse.png b/core/datacap-ui/public/static/images/plugin/clickhouse.png similarity index 100% rename from core/datacap-ui/public/static/images/plugin/ClickHouse.png rename to core/datacap-ui/public/static/images/plugin/clickhouse.png diff --git a/core/datacap-ui/public/static/images/plugin/InfluxDB.png b/core/datacap-ui/public/static/images/plugin/influxdb.png similarity index 100% rename from core/datacap-ui/public/static/images/plugin/InfluxDB.png rename to core/datacap-ui/public/static/images/plugin/influxdb.png diff --git a/core/datacap-ui/public/static/images/plugin/MySQL.png b/core/datacap-ui/public/static/images/plugin/mysql.png similarity index 100% rename from core/datacap-ui/public/static/images/plugin/MySQL.png rename to core/datacap-ui/public/static/images/plugin/mysql.png diff --git a/core/datacap-ui/src/views/pages/admin/source/SourceInfo.vue b/core/datacap-ui/src/views/pages/admin/source/SourceInfo.vue index 73120c4d..a8859222 100644 --- a/core/datacap-ui/src/views/pages/admin/source/SourceInfo.vue +++ b/core/datacap-ui/src/views/pages/admin/source/SourceInfo.vue @@ -15,7 +15,7 @@ :key="plugin.name" :value="plugin.name"> - + diff --git a/logo/executor/seatunnel.svg b/logo/executor/seatunnel.svg index bbaf71f0..63db209e 100644 --- a/logo/executor/seatunnel.svg +++ b/logo/executor/seatunnel.svg @@ -1,11 +1,11 @@ - + - + + + + diff --git a/plugin/datacap-plugin-influxdb/pom.xml b/plugin/datacap-plugin-influxdb/pom.xml index 5203af04..9020d8f6 100644 --- a/plugin/datacap-plugin-influxdb/pom.xml +++ b/plugin/datacap-plugin-influxdb/pom.xml @@ -11,11 +11,7 @@ 4.0.0 datacap-plugin-influxdb - DataCap - InfluxDB - - - datacap-influxdb - + DataCap - Plugin - InfluxDB @@ -32,11 +28,6 @@ kotlin-reflect provided - - org.testcontainers - testcontainers - test - net.suteren.jdbc.influxdb influxdb-jdbc @@ -70,7 +61,24 @@ org.jetbrains.dokka dokka-maven-plugin + + org.apache.maven.plugins + maven-dependency-plugin + + + copy-dependencies + package + + copy-dependencies + + + ${project.build.directory}/dependencies + provided + runtime + + + + - diff --git a/plugin/datacap-plugin-influxdb/src/main/kotlin/io/edurt/datacap/plugin/influxdb/InfluxDBModule.kt b/plugin/datacap-plugin-influxdb/src/main/kotlin/io/edurt/datacap/plugin/influxdb/InfluxDBModule.kt deleted file mode 100644 index e1aa37c4..00000000 --- a/plugin/datacap-plugin-influxdb/src/main/kotlin/io/edurt/datacap/plugin/influxdb/InfluxDBModule.kt +++ /dev/null @@ -1,21 +0,0 @@ -package io.edurt.datacap.plugin.influxdb - -import com.google.inject.multibindings.Multibinder -import io.edurt.datacap.spi.AbstractPluginModule -import io.edurt.datacap.spi.PluginService -import io.edurt.datacap.spi.PluginModule - -class InfluxDBModule : AbstractPluginModule(), PluginModule -{ - override fun get(): AbstractPluginModule - { - return this - } - - override fun configure() - { - Multibinder.newSetBinder(binder(), _root_ide_package_.io.edurt.datacap.spi.PluginService::class.java) - .addBinding() - .to(InfluxDBPlugin::class.java) - } -} diff --git a/plugin/datacap-plugin-influxdb/src/main/kotlin/io/edurt/datacap/plugin/influxdb/InfluxDBPlugin.kt b/plugin/datacap-plugin-influxdb/src/main/kotlin/io/edurt/datacap/plugin/influxdb/InfluxDBPlugin.kt deleted file mode 100644 index 3570524d..00000000 --- a/plugin/datacap-plugin-influxdb/src/main/kotlin/io/edurt/datacap/plugin/influxdb/InfluxDBPlugin.kt +++ /dev/null @@ -1,21 +0,0 @@ -package io.edurt.datacap.plugin.influxdb - -import io.edurt.datacap.spi.PluginService - -class InfluxDBPlugin : _root_ide_package_.io.edurt.datacap.spi.PluginService -{ - override fun validator(): String - { - return "SELECT 1" - } - - override fun driver(): String - { - return "net.suteren.jdbc.influxdb.InfluxDbDriver" - } - - override fun connectType(): String - { - return "influxdb" - } -} diff --git a/plugin/datacap-plugin-influxdb/src/main/kotlin/io/edurt/datacap/plugin/influxdb/InfluxdbPlugin.kt b/plugin/datacap-plugin-influxdb/src/main/kotlin/io/edurt/datacap/plugin/influxdb/InfluxdbPlugin.kt new file mode 100644 index 00000000..612e0279 --- /dev/null +++ b/plugin/datacap-plugin-influxdb/src/main/kotlin/io/edurt/datacap/plugin/influxdb/InfluxdbPlugin.kt @@ -0,0 +1,5 @@ +package io.edurt.datacap.plugin.influxdb + +import io.edurt.datacap.plugin.Plugin + +class InfluxdbPlugin : Plugin() diff --git a/plugin/datacap-plugin-influxdb/src/main/kotlin/io/edurt/datacap/plugin/influxdb/InfluxdbService.kt b/plugin/datacap-plugin-influxdb/src/main/kotlin/io/edurt/datacap/plugin/influxdb/InfluxdbService.kt new file mode 100644 index 00000000..65c17eef --- /dev/null +++ b/plugin/datacap-plugin-influxdb/src/main/kotlin/io/edurt/datacap/plugin/influxdb/InfluxdbService.kt @@ -0,0 +1,79 @@ +package io.edurt.datacap.plugin.influxdb + +import io.edurt.datacap.spi.PluginService +import io.edurt.datacap.spi.model.Configure + +class InfluxdbService : PluginService +{ + override fun validator(): String + { + return "SELECT 1" + } + + override fun driver(): String + { + return "net.suteren.jdbc.influxdb.InfluxDbDriver" + } + + override fun connectType(): String + { + return "influxdb" + } + + override fun url(configure: Configure): String + { + // 如果已经有自定义 URL，直接使用 + // If there is a custom URL, use it + if (configure.url.isPresent) + { + return configure.url.get() + } + + val builder = StringBuilder("jdbc:${configure.type}:") + + // 添加主机和端口 + // Add host and port + builder.append(configure.host) + if (configure.port != null) + { + builder.append(":").append(configure.port) + } + builder.append("/") + + // 添加参数 + // Add parameters + val params = mutableListOf() + + // 添加数据库 + // Add database + configure.database.ifPresent { db -> + params.add("db=$db") + } + + // 添加用户名和密码 + // Add username and password + configure.username.ifPresent { username -> + params.add("u=$username") + } + configure.password.ifPresent { password -> + params.add("p=$password") + } + + // 添加其他环境参数 + // Add other environment parameters + configure.env.ifPresent { envMap -> + envMap.forEach { (key, value) -> + params.add("$key=$value") + } + } + + // 将所有参数添加到 URL + // Add all parameters to the URL + if (params.isNotEmpty()) + { + builder.append("?").append(params.joinToString("&")) + } + + return builder.toString() + } +} diff --git a/plugin/datacap-plugin-influxdb/src/main/resources/META-INF/services/io.edurt.datacap.plugin.Plugin b/plugin/datacap-plugin-influxdb/src/main/resources/META-INF/services/io.edurt.datacap.plugin.Plugin new file mode 100644 index 00000000..a5dd13be --- /dev/null +++ b/plugin/datacap-plugin-influxdb/src/main/resources/META-INF/services/io.edurt.datacap.plugin.Plugin @@ -0,0 +1 @@ +io.edurt.datacap.plugin.influxdb.InfluxdbPlugin \ No newline at end of file diff --git a/plugin/datacap-plugin-influxdb/src/main/resources/META-INF/services/io.edurt.datacap.plugin.Service b/plugin/datacap-plugin-influxdb/src/main/resources/META-INF/services/io.edurt.datacap.plugin.Service new file mode 100644 index 00000000..32aa5f20 --- /dev/null +++ b/plugin/datacap-plugin-influxdb/src/main/resources/META-INF/services/io.edurt.datacap.plugin.Service @@ -0,0 +1 @@ +io.edurt.datacap.plugin.influxdb.InfluxdbService \ No newline at end of file diff --git a/plugin/datacap-plugin-influxdb/src/main/resources/META-INF/services/io.edurt.datacap.spi.PluginModule b/plugin/datacap-plugin-influxdb/src/main/resources/META-INF/services/io.edurt.datacap.spi.PluginModule deleted file mode 100644 index 5ff37409..00000000 --- a/plugin/datacap-plugin-influxdb/src/main/resources/META-INF/services/io.edurt.datacap.spi.PluginModule +++ /dev/null @@ -1 +0,0 @@ -io.edurt.datacap.plugin.influxdb.InfluxDBModule diff --git a/plugin/datacap-plugin-influxdb/src/test/kotlin/io/edurt/datacap/plugin/influxdb/InfluxDBModuleTest.kt b/plugin/datacap-plugin-influxdb/src/test/kotlin/io/edurt/datacap/plugin/influxdb/InfluxDBModuleTest.kt deleted file mode 100644 index 723c9fa3..00000000 --- a/plugin/datacap-plugin-influxdb/src/test/kotlin/io/edurt/datacap/plugin/influxdb/InfluxDBModuleTest.kt +++ /dev/null @@ -1,32 +0,0 @@ -package io.edurt.datacap.plugin.influxdb - -import com.google.inject.Guice.createInjector -import com.google.inject.Injector -import com.google.inject.Key -import com.google.inject.TypeLiteral -import io.edurt.datacap.spi.Plugin -import org.junit.Assert.assertEquals -import org.junit.Before -import org.junit.Test - -class InfluxDBModuleTest -{ - private val name = "InfluxDB" - private var injector: Injector? = null - - @Before - fun before() - { - injector = createInjector(InfluxDBModule()) - } - - @Test - fun test() - { - injector !!.getInstance(Key.get(object : TypeLiteral>() - {})) - .stream() - .findFirst() - .ifPresent { assertEquals(name, it.name()) } - } -} diff --git a/plugin/datacap-plugin-influxdb/src/test/kotlin/io/edurt/datacap/plugin/influxdb/InfluxDBPluginTest.kt b/plugin/datacap-plugin-influxdb/src/test/kotlin/io/edurt/datacap/plugin/influxdb/InfluxDBPluginTest.kt deleted file mode 100644 index e90ad629..00000000 --- a/plugin/datacap-plugin-influxdb/src/test/kotlin/io/edurt/datacap/plugin/influxdb/InfluxDBPluginTest.kt +++ /dev/null @@ -1,80 +0,0 @@ -package io.edurt.datacap.plugin.influxdb - -import com.google.common.collect.Lists -import com.google.inject.Guice.createInjector -import com.google.inject.Injector -import com.google.inject.Key -import com.google.inject.TypeLiteral -import io.edurt.datacap.file.FileManager -import io.edurt.datacap.spi.Plugin -import io.edurt.datacap.spi.model.Configure -import io.edurt.datacap.spi.model.Response -import org.junit.Assert.assertTrue -import org.junit.Before -import org.junit.Test -import org.slf4j.LoggerFactory.getLogger -import org.testcontainers.containers.Network -import org.testcontainers.lifecycle.Startables -import org.testcontainers.shaded.org.awaitility.Awaitility.given -import java.util.* -import java.util.concurrent.TimeUnit - -class InfluxDBPluginTest -{ - private val log = getLogger(this.javaClass) - private val host = "TestInfluxDBContainer" - private var container: InfluxDBContainer? = null - private var injector: Injector? = null - private var configure: Configure? = null - - @Before - fun before() - { - val network = Network.newNetwork() - container = InfluxDBContainer() - .withNetwork(network) - .withNetworkAliases(host) - container?.portBindings = Lists.newArrayList(String.format("%s:%s", InfluxDBContainer.PORT, InfluxDBContainer.DOCKER_PORT)) - Startables.deepStart(java.util.stream.Stream.of(container)) - .join() - log.info("InfluxDB container started") - given().ignoreExceptions() - .await() - .atMost(1, TimeUnit.MINUTES) - - injector = createInjector( - InfluxDBModule(), - FileManager() - ) - configure = Configure() - configure !!.injector = injector - configure !!.host = container?.host - configure !!.port = InfluxDBContainer.PORT - configure !!.username = Optional.of(InfluxDBContainer.USERNAME) - configure !!.password = Optional.of(InfluxDBContainer.PASSWORD) - configure !!.env = Optional.of(mapOf("useEncryption" to false, "useHTTP2" to false, "db" to "test")) - } - - @Test - fun test() - { - injector !!.getInstance(Key.get(object : TypeLiteral>() - {})) - .stream() - .findFirst() - .ifPresent { - it.connect(configure) - val response: Response = it.execute(it.validator()) - log.info("================ Plugin executed information =================") - if (! response.isSuccessful) - { - log.error("Message: {}", response.message) - } - else - { - response.columns.forEach { column -> log.info(column.toString()) } - } - assertTrue(response.isSuccessful) - } - } -} diff --git a/pom.xml b/pom.xml index b100dbf8..12a91989 100644 --- a/pom.xml +++ b/pom.xml @@ -75,7 +75,7 @@ - + plugin/datacap-plugin-influxdb executor/datacap-executor-spi executor/datacap-executor-local executor/datacap-executor-seatunnel diff --git a/test/datacap-test-convert/pom.xml b/test/datacap-test-convert/pom.xml index 31faebb3..7b4f4581 100644 --- a/test/datacap-test-convert/pom.xml +++ b/test/datacap-test-convert/pom.xml @@ -47,6 +47,10 @@ + + org.jetbrains.dokka + dokka-maven-plugin + org.jetbrains.kotlin kotlin-maven-plugin diff --git a/test/datacap-test-core/pom.xml b/test/datacap-test-core/pom.xml index 77812851..1b8b6601 100644 --- a/test/datacap-test-core/pom.xml +++ b/test/datacap-test-core/pom.xml @@ -35,6 +35,10 @@ + + org.jetbrains.dokka + dokka-maven-plugin + org.jetbrains.kotlin kotlin-maven-plugin diff --git a/test/datacap-test-executor/pom.xml b/test/datacap-test-executor/pom.xml index 6d7264a1..23b340a0 100644 --- a/test/datacap-test-executor/pom.xml +++ b/test/datacap-test-executor/pom.xml @@ -53,6 +53,10 @@ + + org.jetbrains.dokka + dokka-maven-plugin + org.jetbrains.kotlin kotlin-maven-plugin diff --git a/test/datacap-test-lib/pom.xml b/test/datacap-test-lib/pom.xml index fc2eabf6..275ebb5e 100644 --- a/test/datacap-test-lib/pom.xml +++ b/test/datacap-test-lib/pom.xml @@ -40,6 +40,10 @@ + + org.jetbrains.dokka + dokka-maven-plugin + org.jetbrains.kotlin kotlin-maven-plugin diff --git a/test/datacap-test-plugin/pom.xml b/test/datacap-test-plugin/pom.xml index c0d0490e..e0c3e9a5 100644 --- a/test/datacap-test-plugin/pom.xml +++ b/test/datacap-test-plugin/pom.xml @@ -19,5 +19,52 @@ datacap-plugin ${project.version} + + org.testcontainers + testcontainers + test + + + io.edurt.datacap + datacap-spi + test + + + io.edurt.datacap + datacap-convert-spi + test + + + io.edurt.datacap + datacap-plugin-influxdb + ${project.version} + test + + + + + + org.jetbrains.dokka + dokka-maven-plugin + + + org.jetbrains.kotlin + kotlin-maven-plugin + + + test-compile + + test-compile + + + + ${project.basedir}/src/test/kotlin + + + + + + + diff --git a/plugin/datacap-plugin-influxdb/src/test/kotlin/io/edurt/datacap/plugin/influxdb/InfluxDBContainer.kt b/test/datacap-test-plugin/src/test/java/io/edurt/datacap/test/influxdb/InfluxdbContainer.kt similarity index 94% rename from plugin/datacap-plugin-influxdb/src/test/kotlin/io/edurt/datacap/plugin/influxdb/InfluxDBContainer.kt rename to test/datacap-test-plugin/src/test/java/io/edurt/datacap/test/influxdb/InfluxdbContainer.kt index 7d75cb6b..70fe9516 100644 --- a/plugin/datacap-plugin-influxdb/src/test/kotlin/io/edurt/datacap/plugin/influxdb/InfluxDBContainer.kt +++ b/test/datacap-test-plugin/src/test/java/io/edurt/datacap/test/influxdb/InfluxdbContainer.kt @@ -4,7 +4,7 @@ import org.testcontainers.containers.GenericContainer import org.testcontainers.utility.DockerImageName import org.testcontainers.utility.DockerImageName.parse -class InfluxDBContainer : GenericContainer +class InfluxdbContainer : GenericContainer { constructor() : super(parse(DEFAULT_IMAGE_NAME)) { diff --git a/test/datacap-test-plugin/src/test/java/io/edurt/datacap/test/influxdb/InfluxdbPluginTest.kt b/test/datacap-test-plugin/src/test/java/io/edurt/datacap/test/influxdb/InfluxdbPluginTest.kt new file mode 100644 index 00000000..3ecb0548 --- /dev/null +++ b/test/datacap-test-plugin/src/test/java/io/edurt/datacap/test/influxdb/InfluxdbPluginTest.kt @@ -0,0 +1,29 @@ +package io.edurt.datacap.plugin.influxdb + +import io.edurt.datacap.plugin.PluginConfigure +import io.edurt.datacap.plugin.PluginManager +import io.edurt.datacap.plugin.utils.PluginPathUtils +import org.junit.Assert.assertNotNull +import org.junit.Test + +class InfluxdbPluginTest +{ + private val pluginManager: PluginManager + private val pluginName = "Influxdb" + + init + { + val projectRoot = PluginPathUtils.findProjectRoot() + val config = PluginConfigure.builder() + .pluginsDir(projectRoot.resolve("plugin/datacap-plugin-influxdb")) + .build() + + pluginManager = PluginManager(config).apply { start() } + } + + @Test + fun test() + { + assertNotNull(pluginManager.getPlugin(pluginName).get()) + } +} diff --git a/test/datacap-test-plugin/src/test/java/io/edurt/datacap/test/influxdb/InfluxdbServiceTest.kt b/test/datacap-test-plugin/src/test/java/io/edurt/datacap/test/influxdb/InfluxdbServiceTest.kt new file mode 100644 index 00000000..7f574515 --- /dev/null +++ b/test/datacap-test-plugin/src/test/java/io/edurt/datacap/test/influxdb/InfluxdbServiceTest.kt @@ -0,0 +1,90 @@ +package io.edurt.datacap.plugin.influxdb + +import edu.umd.cs.findbugs.annotations.SuppressFBWarnings +import io.edurt.datacap.plugin.PluginConfigure +import io.edurt.datacap.plugin.PluginManager +import io.edurt.datacap.plugin.utils.PluginPathUtils +import io.edurt.datacap.spi.PluginService +import io.edurt.datacap.spi.model.Configure +import org.junit.Assert.assertNotNull +import org.junit.Test +import org.slf4j.LoggerFactory.getLogger +import org.testcontainers.containers.Network +import org.testcontainers.lifecycle.Startables +import org.testcontainers.shaded.org.awaitility.Awaitility.given +import java.nio.file.Path +import java.util.* +import java.util.concurrent.TimeUnit + +@SuppressFBWarnings(value = ["RV_RETURN_VALUE_IGNORED_INFERRED", "SA_LOCAL_SELF_ASSIGNMENT"]) +class InfluxdbServiceTest +{ + private val log = getLogger(this.javaClass) + private val pluginManager: PluginManager + private var container: InfluxdbContainer + private val pluginName = "Influxdb" + + init + { + val network = Network.newNetwork() + + container = InfluxdbContainer() + .withNetwork(network) + .withNetworkAliases(this.javaClass.simpleName) + + container.portBindings = listOf(String.format("%s:%s", InfluxdbContainer.PORT, InfluxdbContainer.DOCKER_PORT)) + + Startables.deepStart(java.util.stream.Stream.of(container)) + .join() + + given().ignoreExceptions() + .await() + .atMost(1, TimeUnit.MINUTES) + log.info("Influxdb container started") + + val configFile = this::class + .java + .classLoader + .getResource("influxdb-config.properties") + assertNotNull(configFile) + log.info("Specified config file: {}", configFile) + + val projectRoot = PluginPathUtils.findProjectRoot() + val config = PluginConfigure.builder() + .pluginsDir(projectRoot.resolve(Path.of(configFile.path))) + .build() + + log.info("Initializing plugin manager") + pluginManager = PluginManager(config).apply { start() } + } + + @Test + fun test() + { + val plugin = pluginManager.getPlugin(pluginName).get() + assertNotNull(plugin) + + val configure: Configure = Configure.builder() + .host(container.host) + .port(InfluxdbContainer.PORT) + .username(Optional.of(InfluxdbContainer.USERNAME)) + .password(Optional.of(InfluxdbContainer.PASSWORD)) + .database(Optional.of(InfluxdbContainer.BUCKET)) + .plugin(plugin) + .pluginManager(pluginManager) + .env( + Optional.of( + mapOf( + "useEncryption" to false, + "useHTTP2" to false, + "org" to "TestOrg" + ) + ) + ) + .build() + + val pluginService = plugin.getService(PluginService::class.java) + val response = pluginService.execute(configure, pluginService.validator()) + assertNotNull(response) + } +} diff --git a/test/datacap-test-plugin/src/test/resources/influxdb-config.properties b/test/datacap-test-plugin/src/test/resources/influxdb-config.properties new file mode 100644 index 00000000..382a5869 --- /dev/null +++ b/test/datacap-test-plugin/src/test/resources/influxdb-config.properties @@ -0,0 +1,2 @@ +datacap-plugin-influxdb=plugin/datacap-plugin-influxdb/pom.xml +datacap-convert-json=convert/datacap-convert-json/pom.xml diff --git a/plugin/datacap-plugin-influxdb/src/test/resources/logback.xml b/test/datacap-test-plugin/src/test/resources/logback.xml similarity index 100% rename from plugin/datacap-plugin-influxdb/src/test/resources/logback.xml rename to test/datacap-test-plugin/src/test/resources/logback.xml diff --git a/test/datacap-test-scheduler/pom.xml b/test/datacap-test-scheduler/pom.xml index 9daacf60..b0a80ddc 100644 --- a/test/datacap-test-scheduler/pom.xml +++ b/test/datacap-test-scheduler/pom.xml @@ -47,6 +47,10 @@ + + org.jetbrains.dokka + dokka-maven-plugin + org.jetbrains.kotlin kotlin-maven-plugin