Skip to content

Commit

Permalink
[AMORO-3340]: Database based http session store (apache#3341)
Browse files Browse the repository at this point in the history
* database based http session store

* docs

* spotless
  • Loading branch information
baiyangtx authored Nov 29, 2024
1 parent 9e0f057 commit 604d1c1
Show file tree
Hide file tree
Showing 15 changed files with 384 additions and 138 deletions.
Original file line number Diff line number Diff line change
Expand Up @@ -256,6 +256,12 @@ public class AmoroManagementConf {
.defaultValue("token")
.withDescription("The authentication used by REST APIs, token (default) or basic.");

public static final ConfigOption<Duration> HTTP_SERVER_SESSION_TIMEOUT =
ConfigOptions.key("http-server.session-timeout")
.durationType()
.defaultValue(Duration.ofDays(7))
.withDescription("Timeout for http session.");

public static final ConfigOption<Integer> OPTIMIZING_COMMIT_THREAD_COUNT =
ConfigOptions.key("self-optimizing.commit-thread-count")
.intType()
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -35,6 +35,8 @@
import org.apache.amoro.server.dashboard.utils.CommonUtil;
import org.apache.amoro.server.manager.EventsManager;
import org.apache.amoro.server.manager.MetricManager;
import org.apache.amoro.server.persistence.DataSourceFactory;
import org.apache.amoro.server.persistence.HttpSessionHandlerFactory;
import org.apache.amoro.server.persistence.SqlSessionFactoryProvider;
import org.apache.amoro.server.resource.ContainerMetadata;
import org.apache.amoro.server.resource.OptimizerManager;
Expand Down Expand Up @@ -64,11 +66,12 @@
import org.apache.amoro.utils.JacksonUtil;
import org.apache.commons.lang3.StringUtils;
import org.apache.iceberg.SystemProperties;
import org.eclipse.jetty.server.session.SessionHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.yaml.snakeyaml.Yaml;

import javax.sql.DataSource;

import java.net.InetSocketAddress;
import java.nio.file.Files;
import java.nio.file.Paths;
Expand All @@ -88,6 +91,7 @@ public class AmoroServiceContainer {
public static final String SERVER_CONFIG_FILENAME = "config.yaml";

private final HighAvailabilityContainer haContainer;
private DataSource dataSource;
private DefaultTableService tableService;
private DefaultOptimizingService optimizingService;
private TerminalManager terminalManager;
Expand Down Expand Up @@ -244,7 +248,8 @@ private void initHttpService() {
config -> {
config.addStaticFiles(dashboardServer.configStaticFiles());
config.addStaticFiles("/META-INF/resources/webjars", Location.CLASSPATH);
config.sessionHandler(SessionHandler::new);
config.sessionHandler(
() -> HttpSessionHandlerFactory.createSessionHandler(dataSource, serviceConfig));
config.enableCorsForAllOrigins();
config.jsonMapper(JavalinJsonMapper.createDefaultJsonMapper());
config.showJavalinBanner = false;
Expand Down Expand Up @@ -439,7 +444,8 @@ private void initServiceConfig(Map<String, Object> envConfig) throws Exception {
expandedConfigurationMap.putAll(envConfig);
serviceConfig = Configurations.fromObjectMap(expandedConfigurationMap);
AmoroManagementConfValidator.validateConfig(serviceConfig);
SqlSessionFactoryProvider.getInstance().init(serviceConfig);
dataSource = DataSourceFactory.createDataSource(serviceConfig);
SqlSessionFactoryProvider.getInstance().init(dataSource);
}

private Map<String, Object> initEnvConfig() {
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,141 @@
/*
* 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.amoro.server.persistence;

import org.apache.amoro.config.Configurations;
import org.apache.amoro.server.AmoroManagementConf;
import org.apache.commons.dbcp2.BasicDataSource;
import org.apache.commons.pool2.impl.BaseObjectPoolConfig;
import org.apache.ibatis.jdbc.ScriptRunner;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.sql.DataSource;

import java.io.InputStreamReader;
import java.net.URI;
import java.net.URISyntaxException;
import java.net.URL;
import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.sql.Connection;
import java.sql.ResultSet;
import java.sql.Statement;
import java.time.Duration;

/** Factory to help create data source */
public class DataSourceFactory {
private static final Logger LOG = LoggerFactory.getLogger(DataSourceFactory.class);
private static final String DERBY_INIT_SQL_SCRIPT = "derby/ams-derby-init.sql";
private static final String MYSQL_INIT_SQL_SCRIPT = "mysql/ams-mysql-init.sql";
private static final String POSTGRES_INIT_SQL_SCRIPT = "postgres/ams-postgres-init.sql";

public static DataSource createDataSource(Configurations config) {
BasicDataSource dataSource = new BasicDataSource();
dataSource.setUrl(config.getString(AmoroManagementConf.DB_CONNECTION_URL));
dataSource.setDriverClassName(config.getString(AmoroManagementConf.DB_DRIVER_CLASS_NAME));
String dbType = config.getString(AmoroManagementConf.DB_TYPE);
if (AmoroManagementConf.DB_TYPE_MYSQL.equals(dbType)
|| AmoroManagementConf.DB_TYPE_POSTGRES.equals(dbType)) {
dataSource.setUsername(config.getString(AmoroManagementConf.DB_USER_NAME));
dataSource.setPassword(config.getString(AmoroManagementConf.DB_PASSWORD));
}
dataSource.setDefaultAutoCommit(false);
dataSource.setMaxTotal(config.getInteger(AmoroManagementConf.DB_CONNECT_MAX_TOTAL));
dataSource.setMaxIdle(config.getInteger(AmoroManagementConf.DB_CONNECT_MAX_IDLE));
dataSource.setMinIdle(0);
dataSource.setMaxWaitMillis(config.getLong(AmoroManagementConf.DB_CONNECT_MAX_WAIT_MILLIS));
dataSource.setLogAbandoned(true);
dataSource.setRemoveAbandonedOnBorrow(true);
dataSource.setRemoveAbandonedTimeout(60);
dataSource.setTimeBetweenEvictionRunsMillis(Duration.ofMillis(10 * 60 * 1000L).toMillis());
dataSource.setTestOnBorrow(BaseObjectPoolConfig.DEFAULT_TEST_ON_BORROW);
dataSource.setTestWhileIdle(BaseObjectPoolConfig.DEFAULT_TEST_WHILE_IDLE);
dataSource.setMinEvictableIdleTimeMillis(1000);
dataSource.setNumTestsPerEvictionRun(BaseObjectPoolConfig.DEFAULT_NUM_TESTS_PER_EVICTION_RUN);
dataSource.setTestOnReturn(BaseObjectPoolConfig.DEFAULT_TEST_ON_RETURN);
dataSource.setSoftMinEvictableIdleTimeMillis(
BaseObjectPoolConfig.DEFAULT_SOFT_MIN_EVICTABLE_IDLE_TIME.toMillis());
dataSource.setLifo(BaseObjectPoolConfig.DEFAULT_LIFO);
createTablesIfNeed(dataSource, config);
return dataSource;
}

/** create tables for database */
private static void createTablesIfNeed(DataSource ds, Configurations config) {
boolean initSchema = config.getBoolean(AmoroManagementConf.DB_AUTO_CREATE_TABLES);
if (!initSchema) {
LOG.info("Skip auto create tables due to configuration");
return;
}
String dbTypeConfig = config.getString(AmoroManagementConf.DB_TYPE);
String query = "";
LOG.info("Start create tables, database type:{}", dbTypeConfig);

try (Connection connection = ds.getConnection();
Statement statement = connection.createStatement()) {
if (AmoroManagementConf.DB_TYPE_DERBY.equals(dbTypeConfig)) {
query = "SELECT 1 FROM SYS.SYSTABLES WHERE TABLENAME = 'CATALOG_METADATA'";
} else if (AmoroManagementConf.DB_TYPE_MYSQL.equals(dbTypeConfig)) {
query =
String.format(
"SELECT 1 FROM information_schema.tables WHERE table_schema = '%s' AND table_name = '%s'",
connection.getCatalog(), "catalog_metadata");
} else if (AmoroManagementConf.DB_TYPE_POSTGRES.equals(dbTypeConfig)) {
query =
String.format(
"SELECT 1 FROM information_schema.tables WHERE table_schema = %s AND table_name = '%s'",
"current_schema()", "catalog_metadata");
}
LOG.info("Start check table creation, using query: {}", query);
try (ResultSet rs = statement.executeQuery(query)) {
if (!rs.next()) {
Path script = Paths.get(getInitSqlScriptPath(dbTypeConfig));
LOG.info("Table not exists, start run create tables script file:{}", script);
ScriptRunner runner = new ScriptRunner(connection);
runner.runScript(
new InputStreamReader(Files.newInputStream(script), StandardCharsets.UTF_8));
LOG.info("Tables are created successfully");
} else {
LOG.info("Tables are created, skip auto create tables.");
}
}
} catch (Exception e) {
throw new IllegalStateException("Create tables failed", e);
}
}

private static URI getInitSqlScriptPath(String type) throws URISyntaxException {
String scriptPath = null;
if (type.equals(AmoroManagementConf.DB_TYPE_MYSQL)) {
scriptPath = MYSQL_INIT_SQL_SCRIPT;
} else if (type.equals(AmoroManagementConf.DB_TYPE_DERBY)) {
scriptPath = DERBY_INIT_SQL_SCRIPT;
} else if (type.equals(AmoroManagementConf.DB_TYPE_POSTGRES)) {
scriptPath = POSTGRES_INIT_SQL_SCRIPT;
}
URL scriptUrl = ClassLoader.getSystemResource(scriptPath);
if (scriptUrl == null) {
throw new IllegalStateException("Cannot find init sql script:" + scriptPath);
}
return scriptUrl.toURI();
}
}
Original file line number Diff line number Diff line change
@@ -0,0 +1,90 @@
/*
* 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.amoro.server.persistence;

import org.apache.amoro.config.Configurations;
import org.apache.amoro.server.AmoroManagementConf;
import org.eclipse.jetty.server.session.DatabaseAdaptor;
import org.eclipse.jetty.server.session.DefaultSessionCache;
import org.eclipse.jetty.server.session.JDBCSessionDataStore;
import org.eclipse.jetty.server.session.SessionHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.sql.DataSource;

import java.sql.Connection;
import java.sql.DatabaseMetaData;
import java.sql.SQLException;
import java.time.Duration;

public class HttpSessionHandlerFactory {
private static final Logger LOG = LoggerFactory.getLogger(HttpSessionHandlerFactory.class);

public static SessionHandler createSessionHandler(
DataSource dataSource, Configurations configurations) {
DatabaseAdaptor adaptor = new DatabaseAdaptor();
adaptor.setDatasource(dataSource);

JDBCSessionDataStore dataStore = new JDBCSessionDataStore();
dataStore.setDatabaseAdaptor(adaptor);
dataStore.setSessionTableSchema(new AmoroSessionTableSchema(dataSource));

SessionHandler handler = new SessionHandler();
DefaultSessionCache cache = new DefaultSessionCache(handler);
cache.setSessionDataStore(dataStore);
// set session timeout
Duration sessionTimeout = configurations.get(AmoroManagementConf.HTTP_SERVER_SESSION_TIMEOUT);
handler.setMaxInactiveInterval((int) sessionTimeout.getSeconds());
handler.setSessionCache(cache);
return handler;
}

private static class AmoroSessionTableSchema extends JDBCSessionDataStore.SessionTableSchema {
private final DataSource dataSource;

public AmoroSessionTableSchema(DataSource dataSource) {
setTableName("http_session");
setIdColumn("session_id");
setContextPathColumn("context_path");
setVirtualHostColumn("virtual_host");
setLastNodeColumn("last_node");
setAccessTimeColumn("access_time");
setLastAccessTimeColumn("last_access_time");
setCreateTimeColumn("create_time");
setCookieTimeColumn("cookie_time");
setLastSavedTimeColumn("last_save_time");
setExpiryTimeColumn("expiry_time");
setMaxIntervalColumn("max_interval");
setMapColumn("data_store");
this.dataSource = dataSource;
}

@Override
public void prepareTables() throws SQLException {
LOG.info("Skip jetty prepare http session tables.");
try (Connection connection = dataSource.getConnection()) {
// make the id table
connection.setAutoCommit(true);
DatabaseMetaData metaData = connection.getMetaData();
_dbAdaptor.adaptTo(metaData);
}
}
}
}
Loading

0 comments on commit 604d1c1

Please sign in to comment.