diff --git a/be/src/information_schema/schema_plugins_scanner.cpp b/be/src/information_schema/schema_plugins_scanner.cpp new file mode 100644 index 00000000000000..5f1225f0172361 --- /dev/null +++ b/be/src/information_schema/schema_plugins_scanner.cpp @@ -0,0 +1,150 @@ +// 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. + +#include "information_schema/schema_plugins_scanner.h" + +#include + +#include "core/assert_cast.h" +#include "core/block/block.h" +#include "core/column/column_nullable.h" +#include "core/data_type/data_type_factory.hpp" +#include "core/string_ref.h" +#include "runtime/exec_env.h" +#include "runtime/query_context.h" +#include "runtime/runtime_state.h" +#include "util/client_cache.h" +#include "util/thrift_rpc_helper.h" + +namespace doris { + +std::vector SchemaPluginsScanner::_s_tbls_columns = { + {"PLUGIN_NAME", TYPE_VARCHAR, sizeof(StringRef), true}, + {"PLUGIN_TYPE", TYPE_VARCHAR, sizeof(StringRef), true}, + {"PLUGIN_VERSION", TYPE_VARCHAR, sizeof(StringRef), true}, + {"SOURCE", TYPE_VARCHAR, sizeof(StringRef), true}, + {"DESCRIPTION", TYPE_STRING, sizeof(StringRef), true}, +}; + +SchemaPluginsScanner::SchemaPluginsScanner() + : SchemaScanner(_s_tbls_columns, TSchemaTableType::SCH_PLUGINS) {} + +Status SchemaPluginsScanner::start(RuntimeState* state) { + if (!_is_init) { + return Status::InternalError("used before initialized."); + } + _block_rows_limit = state->batch_size(); + _rpc_timeout_ms = state->execution_timeout() * 1000; + // Plugins are per-FE local state: ask the FE this session is connected to. + _fe_addr = state->get_query_ctx()->current_connect_fe; + return Status::OK(); +} + +Status SchemaPluginsScanner::_get_plugins_block_from_fe() { + TSchemaTableRequestParams schema_table_request_params; + for (int i = 0; i < _s_tbls_columns.size(); i++) { + schema_table_request_params.__isset.columns_name = true; + schema_table_request_params.columns_name.emplace_back(_s_tbls_columns[i].name); + } + schema_table_request_params.__set_current_user_ident(*_param->common_param->current_user_ident); + + TFetchSchemaTableDataRequest request; + request.__set_schema_table_name(TSchemaTableName::PLUGINS); + request.__set_schema_table_params(schema_table_request_params); + + TFetchSchemaTableDataResult result; + RETURN_IF_ERROR(ThriftRpcHelper::rpc( + _fe_addr.hostname, _fe_addr.port, + [&request, &result](FrontendServiceConnection& client) { + client->fetchSchemaTableData(result, request); + }, + _rpc_timeout_ms)); + + Status status(Status::create(result.status)); + if (!status.ok()) { + LOG(WARNING) << "fetch plugins from FE(" << _fe_addr.hostname + << ") failed, errmsg=" << status; + return status; + } + + _plugins_block = Block::create_unique(); + for (int i = 0; i < _s_tbls_columns.size(); ++i) { + auto data_type = + DataTypeFactory::instance().create_data_type(_s_tbls_columns[i].type, true); + _plugins_block->insert(ColumnWithTypeAndName(data_type->create_column(), data_type, + _s_tbls_columns[i].name)); + } + _plugins_block->reserve(_block_rows_limit); + + std::vector result_data = std::move(result.data_batch); + if (!result_data.empty()) { + auto col_size = result_data[0].column_value.size(); + if (col_size != _s_tbls_columns.size()) { + return Status::InternalError("plugins schema is not match for FE and BE"); + } + } + + for (int i = 0; i < result_data.size(); i++) { + const TRow& row = result_data[i]; + for (int j = 0; j < _s_tbls_columns.size(); j++) { + const TCell& cell = row.column_value[j]; + // An unset string cell means SQL NULL (e.g. unknown PLUGIN_VERSION); + // insert_block_column would materialize it as an empty string. + if (!cell.__isset.stringVal) { + MutableColumnPtr mutable_col_ptr = + IColumn::mutate(std::move(_plugins_block->get_by_position(j).column)); + assert_cast(mutable_col_ptr.get())->insert_data(nullptr, 0); + _plugins_block->replace_by_position(j, std::move(mutable_col_ptr)); + continue; + } + RETURN_IF_ERROR( + insert_block_column(cell, j, _plugins_block.get(), _s_tbls_columns[j].type)); + } + } + return Status::OK(); +} + +Status SchemaPluginsScanner::get_next_block_internal(Block* block, bool* eos) { + if (!_is_init) { + return Status::InternalError("Used before initialized."); + } + + if (nullptr == block || nullptr == eos) { + return Status::InternalError("input pointer is nullptr."); + } + + if (_plugins_block == nullptr) { + RETURN_IF_ERROR(_get_plugins_block_from_fe()); + _total_rows = static_cast(_plugins_block->rows()); + } + + if (_row_idx == _total_rows) { + *eos = true; + return Status::OK(); + } + + int current_batch_rows = std::min(_block_rows_limit, _total_rows - _row_idx); + ScopedMutableBlock scoped_mblock(block); + auto& mblock = scoped_mblock.mutable_block(); + RETURN_IF_ERROR(mblock.add_rows(_plugins_block.get(), _row_idx, current_batch_rows)); + _row_idx += current_batch_rows; + + *eos = _row_idx == _total_rows; + return Status::OK(); +} + +} // namespace doris diff --git a/be/src/information_schema/schema_plugins_scanner.h b/be/src/information_schema/schema_plugins_scanner.h new file mode 100644 index 00000000000000..1e336527843cbc --- /dev/null +++ b/be/src/information_schema/schema_plugins_scanner.h @@ -0,0 +1,56 @@ +// 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. + +#pragma once + +#include + +#include "common/status.h" +#include "information_schema/schema_scanner.h" + +namespace doris { +class RuntimeState; +class Block; + +// information_schema.plugins: kernel plugin inventory of the currently +// connected FE (plugins are per-FE local state, so the RPC targets the +// connected FE instead of the master). +class SchemaPluginsScanner : public SchemaScanner { + ENABLE_FACTORY_CREATOR(SchemaPluginsScanner); + +public: + SchemaPluginsScanner(); + ~SchemaPluginsScanner() override = default; + + Status start(RuntimeState* state) override; + Status get_next_block_internal(Block* block, bool* eos) override; + + static std::vector _s_tbls_columns; + +private: + Status _get_plugins_block_from_fe(); + + TNetworkAddress _fe_addr; + + int _block_rows_limit = 4096; + int _row_idx = 0; + int _total_rows = 0; + int _rpc_timeout_ms = 3000; + std::unique_ptr _plugins_block = nullptr; +}; + +} // namespace doris diff --git a/be/src/information_schema/schema_scanner.cpp b/be/src/information_schema/schema_scanner.cpp index b2b38c83d66410..4b4850629eaf8e 100644 --- a/be/src/information_schema/schema_scanner.cpp +++ b/be/src/information_schema/schema_scanner.cpp @@ -66,6 +66,7 @@ #include "information_schema/schema_load_job_scanner.h" #include "information_schema/schema_metadata_name_ids_scanner.h" #include "information_schema/schema_partitions_scanner.h" +#include "information_schema/schema_plugins_scanner.h" #include "information_schema/schema_processlist_scanner.h" #include "information_schema/schema_profiling_scanner.h" #include "information_schema/schema_role_mappings_scanner.h" @@ -286,6 +287,8 @@ std::unique_ptr SchemaScanner::create(TSchemaTableType::type type return SchemaFileCacheInfoScanner::create_unique(); case TSchemaTableType::SCH_AUTHENTICATION_INTEGRATIONS: return SchemaAuthenticationIntegrationsScanner::create_unique(); + case TSchemaTableType::SCH_PLUGINS: + return SchemaPluginsScanner::create_unique(); case TSchemaTableType::SCH_ROLE_MAPPINGS: return SchemaRoleMappingsScanner::create_unique(); case TSchemaTableType::SCH_TABLE_STREAMS: diff --git a/fe/fe-authentication/fe-authentication-handler/src/main/java/org/apache/doris/authentication/handler/AuthenticationPluginManager.java b/fe/fe-authentication/fe-authentication-handler/src/main/java/org/apache/doris/authentication/handler/AuthenticationPluginManager.java index 3bbfe6459ead0e..69f7b019ce3c59 100644 --- a/fe/fe-authentication/fe-authentication-handler/src/main/java/org/apache/doris/authentication/handler/AuthenticationPluginManager.java +++ b/fe/fe-authentication/fe-authentication-handler/src/main/java/org/apache/doris/authentication/handler/AuthenticationPluginManager.java @@ -26,12 +26,11 @@ import org.apache.doris.extension.loader.LoadFailure; import org.apache.doris.extension.loader.LoadReport; import org.apache.doris.extension.loader.PluginHandle; +import org.apache.doris.extension.loader.PluginRegistry; import org.apache.logging.log4j.LogManager; import org.apache.logging.log4j.Logger; -import java.io.Closeable; -import java.io.IOException; import java.nio.file.Path; import java.util.ArrayList; import java.util.Collections; @@ -63,6 +62,9 @@ public class AuthenticationPluginManager { private static final List AUTH_PARENT_FIRST_PREFIXES = Collections.singletonList("org.apache.doris.authentication."); + /** Family label in the process-wide {@link PluginRegistry}. */ + private static final String PLUGIN_FAMILY = "AUTHENTICATION"; + /** Factories by plugin name (e.g., "ldap", "oidc", "password") */ private final Map factories = new ConcurrentHashMap<>(); @@ -88,7 +90,23 @@ public AuthenticationPluginManager(DirectoryPluginRuntimeManager factories.put(factory.name(), factory)); + .forEach(factory -> { + try { + // Snapshot self-reported metadata before publishing the factory so + // one throwing implementation is rejected cleanly. The registry is + // first-wins on (family, name), so re-registration from another + // manager instance is a quiet no-op. First registration also wins + // here, keeping the executable factory and the inventory row the + // same plugin. + PluginRegistry.getInstance().registerBuiltin(PLUGIN_FAMILY, factory); + if (factories.putIfAbsent(factory.name(), factory) != null) { + LOG.warn("Skip duplicated built-in authentication plugin name: {}", factory.name()); + } + } catch (RuntimeException e) { + LOG.warn("Skip built-in authentication plugin factory {}: self-reported metadata failed", + factory.getClass().getName(), e); + } + }); } /** @@ -136,11 +154,14 @@ public void loadAll(List pluginRoots, ClassLoader parent) throws Authentic String pluginName = handle.getPluginName(); AuthenticationPluginFactory existing = factories.putIfAbsent(pluginName, handle.getFactory()); if (existing != null) { - closeClassLoaderQuietly(handle.getClassLoader()); + // Remove the rejected handle from the runtime manager too, so its + // classloader is not retained for the FE lifetime. + runtimeManager.discard(pluginName); LOG.warn("Skip duplicated plugin name: {} from directory {}", pluginName, handle.getPluginDir()); continue; } loadedPlugins++; + PluginRegistry.getInstance().registerExternal(PLUGIN_FAMILY, handle); LOG.info("Loaded external authentication plugin: name={}, pluginDir={}, jarCount={}", pluginName, handle.getPluginDir(), handle.getResolvedJars().size()); } @@ -165,17 +186,6 @@ private static LoadFailure firstNonConflictFailure(List failures) { return null; } - private static void closeClassLoaderQuietly(ClassLoader classLoader) { - if (!(classLoader instanceof Closeable)) { - return; - } - try { - ((Closeable) classLoader).close(); - } catch (IOException ignored) { - // Best effort close. - } - } - /** * Get a factory by plugin name. * diff --git a/fe/fe-authentication/fe-authentication-handler/src/test/java/org/apache/doris/authentication/handler/AuthenticationPluginManagerTest.java b/fe/fe-authentication/fe-authentication-handler/src/test/java/org/apache/doris/authentication/handler/AuthenticationPluginManagerTest.java index 6cde32bb90f0f1..bfb323e0c9542e 100644 --- a/fe/fe-authentication/fe-authentication-handler/src/test/java/org/apache/doris/authentication/handler/AuthenticationPluginManagerTest.java +++ b/fe/fe-authentication/fe-authentication-handler/src/test/java/org/apache/doris/authentication/handler/AuthenticationPluginManagerTest.java @@ -521,6 +521,22 @@ public Optional> get(String pluginName return Optional.empty(); } + @Override + public void discard(String pluginName) { + // This stub bypasses the real loadAll bookkeeping, so honor the + // contract that discarding a reported success closes its classloader. + for (PluginHandle handle : report.getSuccesses()) { + if (handle.getPluginName().equals(pluginName) + && handle.getClassLoader() instanceof Closeable) { + try { + ((Closeable) handle.getClassLoader()).close(); + } catch (IOException ignored) { + // test stub: never expected + } + } + } + } + @Override public List> list() { return Collections.emptyList(); diff --git a/fe/fe-core/src/main/java/org/apache/doris/catalog/SchemaTable.java b/fe/fe-core/src/main/java/org/apache/doris/catalog/SchemaTable.java index fdcb838ef6f685..48d6578bd7b532 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/catalog/SchemaTable.java +++ b/fe/fe-core/src/main/java/org/apache/doris/catalog/SchemaTable.java @@ -862,6 +862,14 @@ public class SchemaTable extends Table { .column("ALTER_USER", ScalarType.createStringType()) .column("MODIFY_TIME", ScalarType.createStringType()) .build())) + .put("plugins", + new SchemaTable(SystemIdGenerator.getNextId(), "plugins", TableType.SCHEMA, + builder().column("PLUGIN_NAME", ScalarType.createVarchar(64)) + .column("PLUGIN_TYPE", ScalarType.createVarchar(64)) + .column("PLUGIN_VERSION", ScalarType.createVarchar(128)) + .column("SOURCE", ScalarType.createVarchar(16)) + .column("DESCRIPTION", ScalarType.createStringType()) + .build())) .put("table_streams", new SchemaTable(SystemIdGenerator.getNextId(), "table_streams", TableType.SCHEMA, builder().column("DB_NAME", ScalarType.createVarchar(NAME_CHAR_LEN)) diff --git a/fe/fe-core/src/main/java/org/apache/doris/connector/ConnectorPluginManager.java b/fe/fe-core/src/main/java/org/apache/doris/connector/ConnectorPluginManager.java index 5735341527030b..a45e0c90051ed2 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/connector/ConnectorPluginManager.java +++ b/fe/fe-core/src/main/java/org/apache/doris/connector/ConnectorPluginManager.java @@ -25,6 +25,7 @@ import org.apache.doris.extension.loader.LoadFailure; import org.apache.doris.extension.loader.LoadReport; import org.apache.doris.extension.loader.PluginHandle; +import org.apache.doris.extension.loader.PluginRegistry; import org.apache.logging.log4j.LogManager; import org.apache.logging.log4j.Logger; @@ -64,6 +65,9 @@ public class ConnectorPluginManager { private static final List CONNECTOR_PARENT_FIRST_PREFIXES = Arrays.asList("org.apache.doris.connector.", "org.apache.doris.filesystem."); + /** Family label in the process-wide {@link PluginRegistry}. */ + private static final String PLUGIN_FAMILY = "CONNECTOR"; + private final List providers = new CopyOnWriteArrayList<>(); private final DirectoryPluginRuntimeManager runtimeManager = new DirectoryPluginRuntimeManager<>(); @@ -74,8 +78,17 @@ public class ConnectorPluginManager { public void loadBuiltins() { ServiceLoader.load(ConnectorProvider.class) .forEach(p -> { - providers.add(p); - LOG.info("Registered built-in connector provider: {}", p.getType()); + try { + // Snapshot self-reported metadata before publishing the provider + // so one throwing implementation is rejected cleanly instead of + // aborting startup or being active without an inventory row. + PluginRegistry.getInstance().registerBuiltin(PLUGIN_FAMILY, p); + providers.add(p); + LOG.info("Registered built-in connector provider: {}", p.getType()); + } catch (RuntimeException e) { + LOG.warn("Skip built-in connector provider {}: self-reported metadata failed", + p.getClass().getName(), e); + } }); } @@ -104,13 +117,31 @@ public void loadPlugins(List pluginRoots) { } for (PluginHandle handle : report.getSuccesses()) { + // Built-ins (and earlier-loaded plugins) must never be displaced by a + // same-name directory jar. + if (hasProviderNamed(handle.getPluginName())) { + LOG.warn("Skip connector plugin '{}' from {}: name conflicts with an already " + + "registered provider", handle.getPluginName(), handle.getPluginDir()); + runtimeManager.discard(handle.getPluginName()); + continue; + } providers.add(handle.getFactory()); + PluginRegistry.getInstance().registerExternal(PLUGIN_FAMILY, handle); LOG.info("Loaded connector plugin: name={}, pluginDir={}, jarCount={}", handle.getPluginName(), handle.getPluginDir(), handle.getResolvedJars().size()); } } + private boolean hasProviderNamed(String name) { + for (ConnectorProvider p : providers) { + if (name.equals(p.name())) { + return true; + } + } + return false; + } + /** * Creates a Connector for the given catalog type by selecting the first supporting provider. * diff --git a/fe/fe-core/src/main/java/org/apache/doris/fs/FileSystemPluginManager.java b/fe/fe-core/src/main/java/org/apache/doris/fs/FileSystemPluginManager.java index 51bad3a55a34d3..db3cbca188384c 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/fs/FileSystemPluginManager.java +++ b/fe/fe-core/src/main/java/org/apache/doris/fs/FileSystemPluginManager.java @@ -23,6 +23,7 @@ import org.apache.doris.extension.loader.LoadFailure; import org.apache.doris.extension.loader.LoadReport; import org.apache.doris.extension.loader.PluginHandle; +import org.apache.doris.extension.loader.PluginRegistry; import org.apache.doris.filesystem.FileSystem; import org.apache.doris.filesystem.spi.FileSystemProvider; @@ -36,6 +37,7 @@ import java.util.List; import java.util.Map; import java.util.ServiceLoader; +import java.util.Set; import java.util.concurrent.CopyOnWriteArrayList; /** @@ -70,6 +72,9 @@ public class FileSystemPluginManager { private static final List FS_PARENT_FIRST_PREFIXES = Arrays.asList("org.apache.doris.filesystem.", "software.amazon.awssdk.", "org.apache.hadoop."); + /** Family label in the process-wide {@link PluginRegistry}. */ + private static final String PLUGIN_FAMILY = "FILESYSTEM"; + private final List providers = new CopyOnWriteArrayList<>(); private final DirectoryPluginRuntimeManager runtimeManager = new DirectoryPluginRuntimeManager<>(); @@ -80,9 +85,20 @@ public class FileSystemPluginManager { public void loadBuiltins() { ServiceLoader.load(FileSystemProvider.class) .forEach(p -> { - providers.add(p); - DatasourcePrintableMap.registerSensitiveKeys(p.sensitivePropertyKeys()); - LOG.info("Registered built-in filesystem provider: {}", p.name()); + try { + // Snapshot all self-reported metadata (sensitive keys included) + // before mutating any store, so one throwing implementation is + // rejected cleanly instead of aborting startup or leaving an + // inventory row for a provider that never became active. + Set sensitiveKeys = p.sensitivePropertyKeys(); + PluginRegistry.getInstance().registerBuiltin(PLUGIN_FAMILY, p); + DatasourcePrintableMap.registerSensitiveKeys(sensitiveKeys); + providers.add(p); + LOG.info("Registered built-in filesystem provider: {}", p.name()); + } catch (RuntimeException e) { + LOG.warn("Skip built-in filesystem provider {}: self-reported metadata failed", + p.getClass().getName(), e); + } }); } @@ -111,15 +127,45 @@ public void loadPlugins(List pluginRoots) { } for (PluginHandle handle : report.getSuccesses()) { + // Built-ins (and earlier-loaded plugins) must never be displaced by a + // same-name directory jar. + if (hasProviderNamed(handle.getPluginName())) { + LOG.warn("Skip filesystem plugin '{}' from {}: name conflicts with an already " + + "registered provider", handle.getPluginName(), handle.getPluginDir()); + runtimeManager.discard(handle.getPluginName()); + continue; + } FileSystemProvider provider = handle.getFactory(); + // Snapshot the self-reported sensitive keys before publishing anything: + // this is plugin code and may throw, and a provider must never be active + // without its inventory row (or vice versa). + Set sensitiveKeys; + try { + sensitiveKeys = provider.sensitivePropertyKeys(); + } catch (RuntimeException | LinkageError e) { + runtimeManager.discard(handle.getPluginName()); + LOG.warn("Skip filesystem plugin '{}' from {}: sensitivePropertyKeys() failed", + handle.getPluginName(), handle.getPluginDir(), e); + continue; + } providers.add(provider); - DatasourcePrintableMap.registerSensitiveKeys(provider.sensitivePropertyKeys()); + DatasourcePrintableMap.registerSensitiveKeys(sensitiveKeys); + PluginRegistry.getInstance().registerExternal(PLUGIN_FAMILY, handle); LOG.info("Loaded filesystem plugin: name={}, pluginDir={}, jarCount={}", handle.getPluginName(), handle.getPluginDir(), handle.getResolvedJars().size()); } } + private boolean hasProviderNamed(String name) { + for (FileSystemProvider p : providers) { + if (name.equals(p.name())) { + return true; + } + } + return false; + } + /** * Creates a FileSystem for the given properties by selecting the first supporting provider. * diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/lineage/LineageEventProcessor.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/lineage/LineageEventProcessor.java index 1f7585634f5714..8a55086d16a5a8 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/nereids/lineage/LineageEventProcessor.java +++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/lineage/LineageEventProcessor.java @@ -23,6 +23,7 @@ import org.apache.doris.extension.loader.LoadFailure; import org.apache.doris.extension.loader.LoadReport; import org.apache.doris.extension.loader.PluginHandle; +import org.apache.doris.extension.loader.PluginRegistry; import org.apache.doris.extension.spi.PluginContext; import org.apache.logging.log4j.LogManager; @@ -65,6 +66,9 @@ public class LineageEventProcessor { private static final Logger LOG = LogManager.getLogger(LineageEventProcessor.class); private static final long EVENT_POLL_TIMEOUT_SECONDS = 5L; + /** Family label in the process-wide {@link PluginRegistry}. */ + private static final String PLUGIN_FAMILY = "LINEAGE"; + /** Parent-first prefixes for child-first classloading isolation. */ private static final List LINEAGE_PARENT_FIRST_PREFIXES = Collections.singletonList("org.apache.doris.nereids.lineage."); @@ -129,6 +133,16 @@ private void discoverPlugins() { factory == null ? "null" : factory.getClass().getName()); continue; } + try { + // Snapshot self-reported metadata before publishing the factory so one + // throwing implementation is rejected cleanly instead of aborting the + // whole built-in discovery or being active without an inventory row. + PluginRegistry.getInstance().registerBuiltin(PLUGIN_FAMILY, factory); + } catch (RuntimeException e) { + LOG.warn("Skip built-in lineage plugin factory {}: self-reported metadata failed", + factory.getClass().getName(), e); + continue; + } LineagePluginFactory existing = factories.putIfAbsent(pluginName, factory); if (existing != null) { LOG.warn("Skip duplicated built-in lineage plugin name: {}", pluginName); @@ -158,9 +172,13 @@ pluginRoots, getClass().getClassLoader(), String pluginName = handle.getPluginName(); LineagePluginFactory existing = factories.putIfAbsent(pluginName, handle.getFactory()); if (existing != null) { + // Remove the rejected handle from the runtime manager too, so its + // classloader is not retained for the FE lifetime. + runtimeManager.discard(pluginName); LOG.warn("Skip duplicated lineage plugin name: {} from directory {}", pluginName, handle.getPluginDir()); } else { + PluginRegistry.getInstance().registerExternal(PLUGIN_FAMILY, handle); LOG.info("Loaded external lineage plugin factory: name={}, pluginDir={}, jarCount={}", pluginName, handle.getPluginDir(), handle.getResolvedJars().size()); } diff --git a/fe/fe-core/src/main/java/org/apache/doris/tablefunction/MetadataGenerator.java b/fe/fe-core/src/main/java/org/apache/doris/tablefunction/MetadataGenerator.java index 1838085cb1f4f2..72acb85e00fa28 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/tablefunction/MetadataGenerator.java +++ b/fe/fe-core/src/main/java/org/apache/doris/tablefunction/MetadataGenerator.java @@ -70,6 +70,7 @@ import org.apache.doris.datasource.maxcompute.MaxComputeExternalCatalog; import org.apache.doris.datasource.metacache.MetaCacheEntryStats; import org.apache.doris.datasource.mvcc.MvccUtil; +import org.apache.doris.extension.loader.PluginRegistry; import org.apache.doris.job.common.JobType; import org.apache.doris.job.extensions.insert.streaming.AbstractStreamingTask; import org.apache.doris.job.extensions.insert.streaming.StreamingInsertJob; @@ -166,6 +167,8 @@ public class MetadataGenerator { private static final ImmutableMap AUTHENTICATION_INTEGRATIONS_COLUMN_TO_INDEX; + private static final ImmutableMap PLUGINS_COLUMN_TO_INDEX; + private static final ImmutableMap ROLE_MAPPINGS_COLUMN_TO_INDEX; private static final ImmutableMap TABLE_STREAMS_COLUMN_TO_INDEX; @@ -259,6 +262,13 @@ public class MetadataGenerator { } AUTHENTICATION_INTEGRATIONS_COLUMN_TO_INDEX = authenticationIntegrationsBuilder.build(); + ImmutableMap.Builder pluginsBuilder = new ImmutableMap.Builder(); + List pluginsColList = SchemaTable.TABLE_MAP.get("plugins").getFullSchema(); + for (int i = 0; i < pluginsColList.size(); i++) { + pluginsBuilder.put(pluginsColList.get(i).getName().toLowerCase(), i); + } + PLUGINS_COLUMN_TO_INDEX = pluginsBuilder.build(); + ImmutableMap.Builder roleMappingsBuilder = new ImmutableMap.Builder(); List roleMappingsColList = SchemaTable.TABLE_MAP.get("role_mappings").getFullSchema(); for (int i = 0; i < roleMappingsColList.size(); i++) { @@ -399,6 +409,10 @@ public static TFetchSchemaTableDataResult getSchemaTableData(TFetchSchemaTableDa result = authenticationIntegrationsMetadataResult(schemaTableParams); columnIndex = AUTHENTICATION_INTEGRATIONS_COLUMN_TO_INDEX; break; + case PLUGINS: + result = pluginsMetadataResult(schemaTableParams); + columnIndex = PLUGINS_COLUMN_TO_INDEX; + break; case ROLE_MAPPINGS: result = roleMappingsMetadataResult(schemaTableParams); columnIndex = ROLE_MAPPINGS_COLUMN_TO_INDEX; @@ -795,6 +809,38 @@ private static TFetchSchemaTableDataResult authenticationIntegrationsMetadataRes return result; } + private static TFetchSchemaTableDataResult pluginsMetadataResult(TSchemaTableRequestParams params) { + if (!params.isSetCurrentUserIdent()) { + return errorResult("current user ident is not set."); + } + UserIdentity currentUserIdentity = UserIdentity.fromThrift(params.getCurrentUserIdent()); + TFetchSchemaTableDataResult result = new TFetchSchemaTableDataResult(); + List dataBatch = Lists.newArrayList(); + result.setDataBatch(dataBatch); + result.setStatus(new TStatus(TStatusCode.OK)); + // The plugin inventory reveals which security/storage components this cluster + // runs; restrict it to ADMIN like SHOW PLUGINS. + if (!Env.getCurrentEnv().getAccessManager().checkGlobalPriv(currentUserIdentity, PrivPredicate.ADMIN)) { + return result; + } + + // Registry rows are load-time snapshots of the current FE; no plugin code runs here. + for (PluginRegistry.PluginRecord record : PluginRegistry.getInstance().list()) { + TRow row = new TRow(); + row.addToColumnValue(new TCell().setStringVal(record.getName())); + row.addToColumnValue(new TCell().setStringVal(record.getType())); + if (record.getVersion() == null) { + row.addToColumnValue(new TCell()); + } else { + row.addToColumnValue(new TCell().setStringVal(record.getVersion())); + } + row.addToColumnValue(new TCell().setStringVal(record.getSource().name())); + row.addToColumnValue(new TCell().setStringVal(record.getDescription())); + dataBatch.add(row); + } + return result; + } + private static TFetchSchemaTableDataResult roleMappingsMetadataResult(TSchemaTableRequestParams params) { if (!params.isSetCurrentUserIdent()) { return errorResult("current user ident is not set."); diff --git a/fe/fe-extension-loader/pom.xml b/fe/fe-extension-loader/pom.xml index 71ca2563ef4046..22ecc4329ea608 100644 --- a/fe/fe-extension-loader/pom.xml +++ b/fe/fe-extension-loader/pom.xml @@ -35,5 +35,15 @@ under the License. fe-extension-spi ${revision} + + + org.apache.logging.log4j + log4j-api + + + org.junit.jupiter + junit-jupiter + test + diff --git a/fe/fe-extension-loader/src/main/java/org/apache/doris/extension/loader/DirectoryPluginRuntimeManager.java b/fe/fe-extension-loader/src/main/java/org/apache/doris/extension/loader/DirectoryPluginRuntimeManager.java index dabcb47cb06ce5..edc2c3abe8f026 100644 --- a/fe/fe-extension-loader/src/main/java/org/apache/doris/extension/loader/DirectoryPluginRuntimeManager.java +++ b/fe/fe-extension-loader/src/main/java/org/apache/doris/extension/loader/DirectoryPluginRuntimeManager.java @@ -22,9 +22,12 @@ import java.io.Closeable; import java.io.IOException; import java.net.MalformedURLException; +import java.net.URISyntaxException; import java.net.URL; import java.nio.file.Files; import java.nio.file.Path; +import java.nio.file.Paths; +import java.security.CodeSource; import java.time.Instant; import java.util.ArrayList; import java.util.Collection; @@ -36,6 +39,9 @@ import java.util.Optional; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; +import java.util.jar.Attributes; +import java.util.jar.JarFile; +import java.util.jar.Manifest; import java.util.stream.Collectors; import java.util.stream.Stream; @@ -151,6 +157,21 @@ public Optional> get(String pluginName) { return Optional.ofNullable(handlesByName.get(pluginName)); } + /** + * Removes a loaded handle and closes its classloader. For callers that reject + * a successfully loaded plugin after the fact (e.g. a family-level name + * conflict with an already registered provider); without this the factory and + * its classloader would stay strongly retained for the FE lifetime. + */ + public void discard(String pluginName) { + synchronized (lifecycleLock) { + PluginHandle handle = handlesByName.remove(pluginName); + if (handle != null) { + closeClassLoader(handle.getClassLoader()); + } + } + } + public List> list() { Collection> handles = handlesByName.values(); List> results = new ArrayList<>(handles); @@ -254,10 +275,12 @@ private PluginHandle loadFromPluginDir(Path pluginDir, ClassLoader parent, Cl throw e; } + // Snapshot self-reported metadata once at load time. A throwing or invalid + // self-report is a load failure; query paths never re-invoke plugin code. String pluginName; try { pluginName = factory.name(); - } catch (RuntimeException e) { + } catch (RuntimeException | LinkageError e) { closeClassLoader(classLoader); throw new PluginLoadException( normalizedDir, @@ -265,22 +288,116 @@ private PluginHandle loadFromPluginDir(Path pluginDir, ClassLoader parent, Cl "Failed to get plugin name from discovered factory in " + normalizedDir, e); } - if (pluginName == null || pluginName.trim().isEmpty()) { + String nameValidationError = PluginNames.validate(pluginName); + if (nameValidationError != null) { closeClassLoader(classLoader); throw new PluginLoadException( normalizedDir, LoadFailure.STAGE_INSTANTIATE, - "Plugin name is empty for directory: " + normalizedDir, + "Invalid plugin name for directory " + normalizedDir + ": " + nameValidationError, null); } + // LinkageError included: a factory jar with a broken classpath (or one + // compiled against a conflicting interface) must fail this plugin only, + // never escape per-plugin isolation. + String description; + try { + description = factory.description(); + } catch (RuntimeException | LinkageError e) { + closeClassLoader(classLoader); + throw new PluginLoadException( + normalizedDir, + LoadFailure.STAGE_INSTANTIATE, + "Failed to get plugin description from discovered factory in " + normalizedDir, + e); + } + + String implementationVersion = readImplementationVersion(factory.getClass(), allJars); + return new PluginHandle<>( pluginName.trim(), normalizedDir, allJars, classLoader, factory, - Instant.now()); + Instant.now(), + description, + implementationVersion); + } + + /** + * Reads Implementation-Version from the MANIFEST of the jar that defined the + * factory class: the class's code source when available (covers layouts where + * the service descriptor sits in a root jar but the implementation lives in + * lib/), otherwise the first candidate jar containing the class entry. + * Version is display-only metadata: failures degrade to null instead of + * failing the load. + */ + private String readImplementationVersion(Class factoryClass, List candidateJars) { + String packagePath = packagePathOf(factoryClass); + Path definingJar = jarOf(factoryClass); + if (definingJar != null) { + try (JarFile jarFile = new JarFile(definingJar.toFile())) { + return manifestImplementationVersion(jarFile, packagePath); + } catch (IOException ignored) { + // Fall through to scanning the candidate jars. + } + } + String classEntry = factoryClass.getName().replace('.', '/') + ".class"; + for (Path jar : candidateJars) { + try (JarFile jarFile = new JarFile(jar.toFile())) { + if (jarFile.getEntry(classEntry) == null) { + continue; + } + return manifestImplementationVersion(jarFile, packagePath); + } catch (IOException ignored) { + // Display-only metadata; fall through to the next candidate jar. + } + } + return null; + } + + private static String manifestImplementationVersion(JarFile jarFile, String packagePath) + throws IOException { + Manifest manifest = jarFile.getManifest(); + if (manifest == null) { + return null; + } + // Per the jar spec a package section ("Name: com/acme/plugin/") overrides + // the main attributes for classes in that package, mirroring + // Package.getImplementationVersion(). + if (packagePath != null) { + Attributes packageAttributes = manifest.getAttributes(packagePath); + if (packageAttributes != null) { + String version = packageAttributes.getValue(Attributes.Name.IMPLEMENTATION_VERSION); + if (version != null) { + return version; + } + } + } + return manifest.getMainAttributes().getValue(Attributes.Name.IMPLEMENTATION_VERSION); + } + + /** Manifest section name for the class's package ("com/acme/plugin/"), or null. */ + private static String packagePathOf(Class clazz) { + String className = clazz.getName(); + int lastDot = className.lastIndexOf('.'); + return lastDot < 0 ? null : className.substring(0, lastDot).replace('.', '/') + "/"; + } + + /** Jar file that defined the class per its protection domain, or null. */ + private static Path jarOf(Class clazz) { + try { + CodeSource codeSource = clazz.getProtectionDomain().getCodeSource(); + if (codeSource == null || codeSource.getLocation() == null) { + return null; + } + Path path = Paths.get(codeSource.getLocation().toURI()); + return Files.isRegularFile(path) ? path : null; + } catch (URISyntaxException | RuntimeException e) { + return null; + } } /** diff --git a/fe/fe-extension-loader/src/main/java/org/apache/doris/extension/loader/PluginHandle.java b/fe/fe-extension-loader/src/main/java/org/apache/doris/extension/loader/PluginHandle.java index 64d5056439a990..e0667c51535ce7 100644 --- a/fe/fe-extension-loader/src/main/java/org/apache/doris/extension/loader/PluginHandle.java +++ b/fe/fe-extension-loader/src/main/java/org/apache/doris/extension/loader/PluginHandle.java @@ -37,9 +37,17 @@ public final class PluginHandle { private final ClassLoader classLoader; private final F factory; private final Instant loadedAt; + private final String description; + private final String implementationVersion; public PluginHandle(String pluginName, Path pluginDir, List resolvedJars, ClassLoader classLoader, F factory, Instant loadedAt) { + this(pluginName, pluginDir, resolvedJars, classLoader, factory, loadedAt, "", null); + } + + public PluginHandle(String pluginName, Path pluginDir, List resolvedJars, + ClassLoader classLoader, F factory, Instant loadedAt, + String description, String implementationVersion) { this.pluginName = requireNonBlank(pluginName, "pluginName"); this.pluginDir = Objects.requireNonNull(pluginDir, "pluginDir"); this.resolvedJars = Collections.unmodifiableList(new ArrayList<>( @@ -47,6 +55,8 @@ public PluginHandle(String pluginName, Path pluginDir, List resolvedJars, this.classLoader = Objects.requireNonNull(classLoader, "classLoader"); this.factory = Objects.requireNonNull(factory, "factory"); this.loadedAt = Objects.requireNonNull(loadedAt, "loadedAt"); + this.description = description == null ? "" : description; + this.implementationVersion = implementationVersion; } public String getPluginName() { @@ -73,6 +83,16 @@ public Instant getLoadedAt() { return loadedAt; } + /** Load-time snapshot of {@code factory.description()}; never re-invokes plugin code. */ + public String getDescription() { + return description; + } + + /** Implementation-Version from the MANIFEST of the jar containing the factory, may be null. */ + public String getImplementationVersion() { + return implementationVersion; + } + private static String requireNonBlank(String value, String fieldName) { Objects.requireNonNull(value, fieldName); String trimmed = value.trim(); diff --git a/fe/fe-extension-loader/src/main/java/org/apache/doris/extension/loader/PluginNames.java b/fe/fe-extension-loader/src/main/java/org/apache/doris/extension/loader/PluginNames.java new file mode 100644 index 00000000000000..08ddf3134dd3f8 --- /dev/null +++ b/fe/fe-extension-loader/src/main/java/org/apache/doris/extension/loader/PluginNames.java @@ -0,0 +1,62 @@ +// 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.doris.extension.loader; + +/** + * Validation rules for self-reported plugin names. + * + *

A plugin name is a configuration-facing identity key, so it is validated + * at load time: non-blank after trimming, bounded length, restricted charset. + * An invalid name is treated as a load failure by the loading side. + */ +final class PluginNames { + + static final int MAX_LENGTH = 64; + + private PluginNames() { + } + + /** + * Validates a self-reported plugin name. + * + * @param name raw value returned by {@code PluginFactory.name()} + * @return null if valid, otherwise a human-readable rejection reason + */ + static String validate(String name) { + if (name == null) { + return "plugin name is null"; + } + String trimmed = name.trim(); + if (trimmed.isEmpty()) { + return "plugin name is blank"; + } + if (trimmed.length() > MAX_LENGTH) { + return "plugin name exceeds " + MAX_LENGTH + " characters"; + } + for (int i = 0; i < trimmed.length(); i++) { + char c = trimmed.charAt(i); + boolean allowed = (c >= 'a' && c <= 'z') || (c >= 'A' && c <= 'Z') || (c >= '0' && c <= '9') + || c == '.' || c == '_' || c == '-'; + if (!allowed) { + return "plugin name contains illegal character '" + c + + "', allowed: letters, digits, '.', '_', '-'"; + } + } + return null; + } +} diff --git a/fe/fe-extension-loader/src/main/java/org/apache/doris/extension/loader/PluginRegistry.java b/fe/fe-extension-loader/src/main/java/org/apache/doris/extension/loader/PluginRegistry.java new file mode 100644 index 00000000000000..4fc736bd8a043e --- /dev/null +++ b/fe/fe-extension-loader/src/main/java/org/apache/doris/extension/loader/PluginRegistry.java @@ -0,0 +1,197 @@ +// 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.doris.extension.loader; + +import org.apache.doris.extension.spi.PluginFactory; + +import org.apache.logging.log4j.LogManager; +import org.apache.logging.log4j.Logger; + +import java.time.Instant; +import java.util.ArrayList; +import java.util.Comparator; +import java.util.List; +import java.util.Objects; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; + +/** + * Process-wide registry of loaded plugins across all plugin families + * (filesystem, connector, authentication, lineage, ...). + * + *

This is the single fact source behind {@code information_schema.plugins}. + * It only stores load-time snapshots (plain strings); listing the registry + * never executes plugin code. + * + *

Rules: + *

    + *
  • Primary key is {@code (type, name)}. The first registration wins; + * later registrations with the same key are rejected, never silently + * overridden. Family managers register built-ins before external plugins, + * so a directory jar can never displace a built-in row.
    • + *
  • Rows only exist for successfully loaded plugins. Load failures are + * reported via logs by the loading side and never enter the registry.
    • + *
+ */ +public final class PluginRegistry { + + /** Where a plugin was loaded from. */ + public enum PluginSource { + /** Bundled on the FE classpath, discovered via ServiceLoader. */ + BUILTIN, + /** Loaded from a plugin directory jar. */ + EXTERNAL + } + + /** Immutable load-time snapshot of one plugin. */ + public static final class PluginRecord { + private final String type; + private final String name; + private final String version; + private final String description; + private final PluginSource source; + private final Instant loadTime; + + public PluginRecord(String type, String name, String version, String description, + PluginSource source, Instant loadTime) { + this.type = Objects.requireNonNull(type, "type"); + this.name = Objects.requireNonNull(name, "name"); + this.version = version; + this.description = description == null ? "" : description; + this.source = Objects.requireNonNull(source, "source"); + this.loadTime = Objects.requireNonNull(loadTime, "loadTime"); + } + + public String getType() { + return type; + } + + public String getName() { + return name; + } + + /** Plugin release version from jar MANIFEST Implementation-Version, may be null. */ + public String getVersion() { + return version; + } + + public String getDescription() { + return description; + } + + public PluginSource getSource() { + return source; + } + + public Instant getLoadTime() { + return loadTime; + } + } + + private static final Logger LOG = LogManager.getLogger(PluginRegistry.class); + + private static final PluginRegistry INSTANCE = new PluginRegistry(); + + private final ConcurrentMap recordsByKey = new ConcurrentHashMap<>(); + + private PluginRegistry() { + } + + public static PluginRegistry getInstance() { + return INSTANCE; + } + + /** + * Registers a load-time snapshot for one plugin. + * + *

Rejects (returns false) when the name is invalid or the + * {@code (type, name)} key is already taken. Rejection never throws so a + * bad self-reported name cannot break FE startup paths that register + * built-ins; callers decide how to react. + * + * @param type plugin family label, e.g. "FILESYSTEM", "AUTHENTICATION" + * @param name plugin identity within the family + * @param version plugin release version, may be null when unknown + * @param description one-line description, may be null + * @param source BUILTIN or EXTERNAL + * @return true if the record was inserted, false if rejected + */ + public boolean register(String type, String name, String version, String description, PluginSource source) { + String validationError = PluginNames.validate(name); + if (validationError != null) { + LOG.warn("Reject plugin registration with invalid name: type={}, name={}, reason={}", + type, name, validationError); + return false; + } + PluginRecord record = new PluginRecord(type, name.trim(), version, description, source, Instant.now()); + PluginRecord existing = recordsByKey.putIfAbsent(key(record.getType(), record.getName()), record); + if (existing != null) { + // The same plugin may be legitimately loaded by more than one manager + // instance within a family (e.g. authentication); keep the first row. + LOG.info("Skip duplicated plugin registration: type={}, name={}, existingSource={}, newSource={}", + record.getType(), record.getName(), existing.getSource(), source); + return false; + } + LOG.info("Registered plugin: type={}, name={}, version={}, source={}", + record.getType(), record.getName(), version, source); + return true; + } + + /** + * Registers a classpath built-in factory. Snapshots {@code name()} and + * {@code description()} now; the version comes from the Implementation-Version + * of the jar that bundles the factory class, when available. + */ + public boolean registerBuiltin(String type, PluginFactory factory) { + return register(type, factory.name(), implementationVersionOf(factory.getClass()), + factory.description(), PluginSource.BUILTIN); + } + + /** Registers a directory-loaded plugin from its load-time handle snapshot. */ + public boolean registerExternal(String type, PluginHandle handle) { + return register(type, handle.getPluginName(), handle.getImplementationVersion(), + handle.getDescription(), PluginSource.EXTERNAL); + } + + /** Returns a snapshot of all registered plugins sorted by (type, name). */ + public List list() { + List records = new ArrayList<>(recordsByKey.values()); + records.sort(Comparator.comparing(PluginRecord::getType).thenComparing(PluginRecord::getName)); + return records; + } + + /** Removes all records. Only for tests. */ + public void clearForTest() { + recordsByKey.clear(); + } + + /** + * Best-effort release version of a classpath (built-in) plugin, from the + * Implementation-Version of the jar that contains the given class. + * + * @return the version, or null when unavailable (e.g. classes directory) + */ + public static String implementationVersionOf(Class clazz) { + Package pkg = clazz.getPackage(); + return pkg == null ? null : pkg.getImplementationVersion(); + } + + private static String key(String type, String name) { + return type + " " + name; + } +} diff --git a/fe/fe-extension-loader/src/test/java/org/apache/doris/extension/loader/DirectoryPluginRuntimeManagerMetadataTest.java b/fe/fe-extension-loader/src/test/java/org/apache/doris/extension/loader/DirectoryPluginRuntimeManagerMetadataTest.java new file mode 100644 index 00000000000000..658ec773e4155d --- /dev/null +++ b/fe/fe-extension-loader/src/test/java/org/apache/doris/extension/loader/DirectoryPluginRuntimeManagerMetadataTest.java @@ -0,0 +1,143 @@ +// 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.doris.extension.loader; + +import org.apache.doris.extension.loader.testplugins.BadNameTestPluginFactory; +import org.apache.doris.extension.loader.testplugins.MetadataTestPluginFactory; +import org.apache.doris.extension.spi.PluginFactory; + +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +import java.io.IOException; +import java.io.InputStream; +import java.nio.charset.StandardCharsets; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.Collections; +import java.util.List; +import java.util.jar.Attributes; +import java.util.jar.JarEntry; +import java.util.jar.JarOutputStream; +import java.util.jar.Manifest; + +/** + * Verifies load-time metadata snapshotting: MANIFEST Implementation-Version, + * description snapshot, and self-reported name validation. + */ +class DirectoryPluginRuntimeManagerMetadataTest { + + @TempDir + Path tempDir; + + @Test + void testManifestVersionAndDescriptionSnapshot() throws IOException { + Path root = tempDir.resolve("root"); + createPluginJar(root.resolve("metadata-test").resolve("metadata-test.jar"), + MetadataTestPluginFactory.class, "3.1.4"); + + DirectoryPluginRuntimeManager manager = new DirectoryPluginRuntimeManager<>(); + LoadReport report = manager.loadAll( + Collections.singletonList(root), + Thread.currentThread().getContextClassLoader(), + PluginFactory.class, + null); + + Assertions.assertEquals(1, report.getSuccesses().size(), () -> failures(report)); + PluginHandle handle = report.getSuccesses().get(0); + Assertions.assertEquals("metadata-test", handle.getPluginName()); + Assertions.assertEquals("Metadata snapshot test plugin", handle.getDescription()); + Assertions.assertEquals("3.1.4", handle.getImplementationVersion()); + } + + @Test + void testMissingManifestVersionDegradesToNull() throws IOException { + Path root = tempDir.resolve("root-no-version"); + createPluginJar(root.resolve("metadata-test").resolve("metadata-test.jar"), + MetadataTestPluginFactory.class, null); + + DirectoryPluginRuntimeManager manager = new DirectoryPluginRuntimeManager<>(); + LoadReport report = manager.loadAll( + Collections.singletonList(root), + Thread.currentThread().getContextClassLoader(), + PluginFactory.class, + null); + + Assertions.assertEquals(1, report.getSuccesses().size(), () -> failures(report)); + Assertions.assertNull(report.getSuccesses().get(0).getImplementationVersion()); + } + + @Test + void testInvalidSelfReportedNameIsLoadFailure() throws IOException { + Path root = tempDir.resolve("root-bad-name"); + createPluginJar(root.resolve("bad-name").resolve("bad-name.jar"), + BadNameTestPluginFactory.class, "1.0"); + + DirectoryPluginRuntimeManager manager = new DirectoryPluginRuntimeManager<>(); + LoadReport report = manager.loadAll( + Collections.singletonList(root), + Thread.currentThread().getContextClassLoader(), + PluginFactory.class, + null); + + Assertions.assertTrue(report.getSuccesses().isEmpty()); + Assertions.assertEquals(1, report.getFailures().size()); + Assertions.assertTrue(report.getFailures().get(0).getMessage().contains("Invalid plugin name"), + () -> report.getFailures().get(0).getMessage()); + } + + /** + * Builds a plugin jar containing the factory's class bytes, a ServiceLoader + * registration, and (optionally) a MANIFEST Implementation-Version. + */ + private static void createPluginJar(Path jarPath, Class factoryClass, + String implementationVersion) throws IOException { + Files.createDirectories(jarPath.getParent()); + Manifest manifest = new Manifest(); + manifest.getMainAttributes().put(Attributes.Name.MANIFEST_VERSION, "1.0"); + if (implementationVersion != null) { + manifest.getMainAttributes().put(Attributes.Name.IMPLEMENTATION_VERSION, implementationVersion); + } + String classEntry = factoryClass.getName().replace('.', '/') + ".class"; + try (JarOutputStream jar = new JarOutputStream(Files.newOutputStream(jarPath), manifest)) { + jar.putNextEntry(new JarEntry(classEntry)); + try (InputStream classBytes = factoryClass.getClassLoader().getResourceAsStream(classEntry)) { + Assertions.assertNotNull(classBytes, "class bytes not found: " + classEntry); + byte[] buffer = new byte[8192]; + int read; + while ((read = classBytes.read(buffer)) != -1) { + jar.write(buffer, 0, read); + } + } + jar.closeEntry(); + jar.putNextEntry(new JarEntry("META-INF/services/" + PluginFactory.class.getName())); + jar.write((factoryClass.getName() + "\n").getBytes(StandardCharsets.UTF_8)); + jar.closeEntry(); + } + } + + private static String failures(LoadReport report) { + StringBuilder sb = new StringBuilder("load failures:"); + List failures = report.getFailures(); + for (LoadFailure failure : failures) { + sb.append(" [").append(failure.getStage()).append("] ").append(failure.getMessage()); + } + return sb.toString(); + } +} diff --git a/fe/fe-extension-loader/src/test/java/org/apache/doris/extension/loader/PluginRegistryTest.java b/fe/fe-extension-loader/src/test/java/org/apache/doris/extension/loader/PluginRegistryTest.java new file mode 100644 index 00000000000000..716a8551eb6d94 --- /dev/null +++ b/fe/fe-extension-loader/src/test/java/org/apache/doris/extension/loader/PluginRegistryTest.java @@ -0,0 +1,145 @@ +// 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.doris.extension.loader; + +import org.apache.doris.extension.loader.PluginRegistry.PluginRecord; +import org.apache.doris.extension.loader.PluginRegistry.PluginSource; +import org.apache.doris.extension.spi.Plugin; +import org.apache.doris.extension.spi.PluginFactory; + +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +import java.util.List; + +class PluginRegistryTest { + + @BeforeEach + void setUp() { + PluginRegistry.getInstance().clearForTest(); + } + + @AfterEach + void tearDown() { + PluginRegistry.getInstance().clearForTest(); + } + + @Test + void testRegisterAndList() { + Assertions.assertTrue(PluginRegistry.getInstance() + .register("AUTHENTICATION", "ldap", "1.2.0", "LDAP authentication", PluginSource.EXTERNAL)); + + List records = PluginRegistry.getInstance().list(); + Assertions.assertEquals(1, records.size()); + PluginRecord record = records.get(0); + Assertions.assertEquals("AUTHENTICATION", record.getType()); + Assertions.assertEquals("ldap", record.getName()); + Assertions.assertEquals("1.2.0", record.getVersion()); + Assertions.assertEquals("LDAP authentication", record.getDescription()); + Assertions.assertEquals(PluginSource.EXTERNAL, record.getSource()); + Assertions.assertNotNull(record.getLoadTime()); + } + + @Test + void testDuplicateKeyRejectedFirstWins() { + Assertions.assertTrue(PluginRegistry.getInstance() + .register("AUTHORIZATION", "ranger", "1.0", "builtin", PluginSource.BUILTIN)); + Assertions.assertFalse(PluginRegistry.getInstance() + .register("AUTHORIZATION", "ranger", "9.9", "impostor", PluginSource.EXTERNAL)); + + List records = PluginRegistry.getInstance().list(); + Assertions.assertEquals(1, records.size()); + Assertions.assertEquals(PluginSource.BUILTIN, records.get(0).getSource()); + Assertions.assertEquals("1.0", records.get(0).getVersion()); + } + + @Test + void testSameNameAcrossFamiliesIsLegal() { + Assertions.assertTrue(PluginRegistry.getInstance() + .register("AUTHENTICATION", "ranger", null, "", PluginSource.EXTERNAL)); + Assertions.assertTrue(PluginRegistry.getInstance() + .register("AUTHORIZATION", "ranger", null, "", PluginSource.EXTERNAL)); + + Assertions.assertEquals(2, PluginRegistry.getInstance().list().size()); + } + + @Test + void testInvalidNameRejected() { + PluginRegistry registry = PluginRegistry.getInstance(); + Assertions.assertFalse(registry.register("FILESYSTEM", null, null, "", PluginSource.BUILTIN)); + Assertions.assertFalse(registry.register("FILESYSTEM", " ", null, "", PluginSource.BUILTIN)); + Assertions.assertFalse(registry.register("FILESYSTEM", "bad name", null, "", PluginSource.BUILTIN)); + StringBuilder tooLong = new StringBuilder(); + for (int i = 0; i < 65; i++) { + tooLong.append('a'); + } + Assertions.assertFalse(registry.register("FILESYSTEM", tooLong.toString(), null, "", PluginSource.BUILTIN)); + Assertions.assertTrue(registry.list().isEmpty()); + } + + @Test + void testNameTrimmedBeforeKeying() { + Assertions.assertTrue(PluginRegistry.getInstance() + .register("CONNECTOR", " hive ", null, "", PluginSource.BUILTIN)); + Assertions.assertFalse(PluginRegistry.getInstance() + .register("CONNECTOR", "hive", null, "", PluginSource.EXTERNAL)); + Assertions.assertEquals("hive", PluginRegistry.getInstance().list().get(0).getName()); + } + + @Test + void testListSortedByTypeThenName() { + PluginRegistry registry = PluginRegistry.getInstance(); + registry.register("FILESYSTEM", "s3", null, "", PluginSource.BUILTIN); + registry.register("AUTHENTICATION", "oidc", null, "", PluginSource.BUILTIN); + registry.register("AUTHENTICATION", "ldap", null, "", PluginSource.BUILTIN); + + List records = registry.list(); + Assertions.assertEquals(3, records.size()); + Assertions.assertEquals("ldap", records.get(0).getName()); + Assertions.assertEquals("oidc", records.get(1).getName()); + Assertions.assertEquals("s3", records.get(2).getName()); + } + + @Test + void testRegisterBuiltinSnapshotsFactoryMetadata() { + Assertions.assertTrue(PluginRegistry.getInstance().registerBuiltin("LINEAGE", new PluginFactory() { + @Override + public String name() { + return "console"; + } + + @Override + public String description() { + return "Console lineage sink"; + } + + @Override + public Plugin create() { + return null; + } + })); + + PluginRecord record = PluginRegistry.getInstance().list().get(0); + Assertions.assertEquals("LINEAGE", record.getType()); + Assertions.assertEquals("console", record.getName()); + Assertions.assertEquals("Console lineage sink", record.getDescription()); + Assertions.assertEquals(PluginSource.BUILTIN, record.getSource()); + } +} diff --git a/fe/fe-extension-loader/src/test/java/org/apache/doris/extension/loader/testplugins/BadNameTestPluginFactory.java b/fe/fe-extension-loader/src/test/java/org/apache/doris/extension/loader/testplugins/BadNameTestPluginFactory.java new file mode 100644 index 00000000000000..4676338002dc48 --- /dev/null +++ b/fe/fe-extension-loader/src/test/java/org/apache/doris/extension/loader/testplugins/BadNameTestPluginFactory.java @@ -0,0 +1,38 @@ +// 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.doris.extension.loader.testplugins; + +import org.apache.doris.extension.spi.Plugin; +import org.apache.doris.extension.spi.PluginFactory; + +/** + * Test factory whose self-reported name violates the load-time charset rule. + */ +public class BadNameTestPluginFactory implements PluginFactory { + + @Override + public String name() { + return "bad name!"; + } + + @Override + public Plugin create() { + return new Plugin() { + }; + } +} diff --git a/fe/fe-extension-loader/src/test/java/org/apache/doris/extension/loader/testplugins/MetadataTestPluginFactory.java b/fe/fe-extension-loader/src/test/java/org/apache/doris/extension/loader/testplugins/MetadataTestPluginFactory.java new file mode 100644 index 00000000000000..0abf8ad95349f8 --- /dev/null +++ b/fe/fe-extension-loader/src/test/java/org/apache/doris/extension/loader/testplugins/MetadataTestPluginFactory.java @@ -0,0 +1,44 @@ +// 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.doris.extension.loader.testplugins; + +import org.apache.doris.extension.spi.Plugin; +import org.apache.doris.extension.spi.PluginFactory; + +/** + * Well-behaved test factory; its class bytes are copied into fabricated plugin + * jars by loader tests. + */ +public class MetadataTestPluginFactory implements PluginFactory { + + @Override + public String name() { + return "metadata-test"; + } + + @Override + public String description() { + return "Metadata snapshot test plugin"; + } + + @Override + public Plugin create() { + return new Plugin() { + }; + } +} diff --git a/fe/fe-extension-spi/src/main/java/org/apache/doris/extension/spi/PluginFactory.java b/fe/fe-extension-spi/src/main/java/org/apache/doris/extension/spi/PluginFactory.java index 5e915b6f83fa72..b80d0dcfd19823 100644 --- a/fe/fe-extension-spi/src/main/java/org/apache/doris/extension/spi/PluginFactory.java +++ b/fe/fe-extension-spi/src/main/java/org/apache/doris/extension/spi/PluginFactory.java @@ -31,6 +31,17 @@ public interface PluginFactory { */ String name(); + /** + * Returns a one-line human-readable description of the plugin. + * + *

Snapshotted once at load time by the loader; never invoked at query time.

+ * + * @return plugin description, empty by default + */ + default String description() { + return ""; + } + /** * Create a new plugin instance. * diff --git a/gensrc/thrift/FrontendService.thrift b/gensrc/thrift/FrontendService.thrift index d87d7df56edd92..dab36a61f60e33 100644 --- a/gensrc/thrift/FrontendService.thrift +++ b/gensrc/thrift/FrontendService.thrift @@ -914,6 +914,7 @@ enum TSchemaTableName { TABLE_STREAMS = 15, TABLE_STREAM_CONSUMPTION = 16, ROLE_MAPPINGS = 17, + PLUGINS = 18, } struct TMetadataTableRequestParams { diff --git a/regression-test/suites/query_p0/schema_table/test_plugins_schema.groovy b/regression-test/suites/query_p0/schema_table/test_plugins_schema.groovy new file mode 100644 index 00000000000000..de2956fca81427 --- /dev/null +++ b/regression-test/suites/query_p0/schema_table/test_plugins_schema.groovy @@ -0,0 +1,76 @@ +// 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. + +import org.junit.Assert + +suite("test_plugins_schema", "p0") { + // Schema check: fixed five columns. + def schema = sql "DESC information_schema.plugins" + def columnNames = schema.collect { it[0] } + Assert.assertEquals( + ["PLUGIN_NAME", "PLUGIN_TYPE", "PLUGIN_VERSION", "SOURCE", "DESCRIPTION"], columnNames) + + // As an admin user: loaded plugins must be listed. Do not require BUILTIN + // rows: the packaged cluster deploys filesystem/connector providers as + // directory plugins (source = EXTERNAL), so a correct inventory may contain + // no BUILTIN row at all. + def rows = sql """ + SELECT PLUGIN_NAME, PLUGIN_TYPE, SOURCE, PLUGIN_VERSION + FROM information_schema.plugins + ORDER BY PLUGIN_TYPE, PLUGIN_NAME + """ + Assert.assertTrue("expect at least one plugin row", rows.size() > 0) + rows.each { row -> + Assert.assertTrue(row[2] == "BUILTIN" || row[2] == "EXTERNAL") + // Unknown versions must surface as SQL NULL, never as an empty string. + Assert.assertTrue("PLUGIN_VERSION must be NULL or non-empty, got ''", + row[3] == null || row[3].toString().length() > 0) + } + + // (type, name) is the primary key: no duplicates may appear. + def keys = rows.collect { "${it[1]}|${it[0]}".toString() } + Assert.assertEquals(keys.size(), keys.unique(false).size()) + + // Filter by family works. + def fsRows = sql """ + SELECT PLUGIN_NAME FROM information_schema.plugins WHERE PLUGIN_TYPE = 'FILESYSTEM' + """ + Assert.assertTrue("expect built-in filesystem providers", fsRows.size() > 0) + + // Non-admin users see an empty inventory (ADMIN-level metadata). + String user = "test_plugins_schema_user" + String pwd = "C123_567p" + try_sql("DROP USER IF EXISTS ${user}") + sql "CREATE USER ${user} IDENTIFIED BY '${pwd}'" + // The JDBC URL carries a default database; without a grant on it the + // connection itself is refused before the query runs. + sql "GRANT SELECT_PRIV ON regression_test TO ${user}" + if (isCloudMode()) { + def clusters = sql " SHOW CLUSTERS; " + Assert.assertTrue(!clusters.isEmpty()) + def validCluster = clusters[0][0] + sql """GRANT USAGE_PRIV ON CLUSTER `${validCluster}` TO ${user}""" + } + try { + connect(user, pwd, context.config.jdbcUrl) { + def result = sql "SELECT * FROM information_schema.plugins" + Assert.assertEquals(0, result.size()) + } + } finally { + try_sql("DROP USER IF EXISTS ${user}") + } +}