Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

[GOBBLIN-1737] Fix bug when using mysql user quota manager #3595

Merged
merged 4 commits into from
Nov 8, 2022
Merged

[GOBBLIN-1737] Fix bug when using mysql user quota manager #3595

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
aad6546
address comments
Mar 25, 2022
28192df
use connectionmanager when httpclient is not cloesable
Mar 25, 2022
b4d8408
[GOBBLIN-1737] Fix bug when using mysql user quota manager
Oct 31, 2022
9e7c65a
fix tests
Nov 1, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 0 additions & 1 deletion ...vice/src/main/java/org/apache/gobblin/service/modules/core/GobblinServiceGuiceModule.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,7 +22,6 @@
import org.apache.gobblin.runtime.api.DagActionStore;
import org.apache.gobblin.runtime.dag_action_store.MysqlDagActionStore;
import org.apache.gobblin.service.modules.orchestration.UserQuotaManager;
//import org.apache.gobblin.service.modules.restli.GobblinServiceFlowConfigV2ResourceHandlerWithWarmStandby;
import org.apache.gobblin.service.modules.restli.GobblinServiceFlowConfigV2ResourceHandlerWithWarmStandby;
import org.apache.gobblin.service.modules.restli.GobblinServiceFlowExecutionResourceHandlerWithWarmStandby;
import org.apache.gobblin.service.monitoring.DagActionStoreChangeMonitor;
Expand Down
15 changes: 10 additions & 5 deletions ...vice/src/main/java/org/apache/gobblin/service/modules/flow/BaseFlowToJobSpecCompiler.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,10 +24,10 @@
import java.util.Map;
import java.util.Properties;

import javax.inject.Inject;
import org.apache.commons.lang3.StringUtils;
import org.apache.gobblin.runtime.spec_catalog.FlowCatalog;
import org.apache.gobblin.service.modules.orchestration.UserQuotaManager;
import org.apache.gobblin.util.reflection.GobblinConstructorUtils;
import org.quartz.CronExpression;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
Expand Down Expand Up @@ -95,8 +95,7 @@ public abstract class BaseFlowToJobSpecCompiler implements SpecCompiler {

private boolean warmStandbyEnabled;

@Inject
UserQuotaManager userQuotaManager;
private Optional<UserQuotaManager> userQuotaManager;

public BaseFlowToJobSpecCompiler(Config config){
this(config,true);
Expand Down Expand Up @@ -128,6 +127,12 @@ public BaseFlowToJobSpecCompiler(Config config, Optional<Logger> log, boolean in
}

this.warmStandbyEnabled = ConfigUtils.getBoolean(config, ServiceConfigKeys.GOBBLIN_SERVICE_WARM_STANDBY_ENABLED_KEY, false);
if (this.warmStandbyEnabled) {
userQuotaManager = Optional.of(GobblinConstructorUtils.invokeConstructor(UserQuotaManager.class,
ConfigUtils.getString(config, ServiceConfigKeys.QUOTA_MANAGER_CLASS, ServiceConfigKeys.DEFAULT_QUOTA_MANAGER), config));
} else {
userQuotaManager = Optional.absent();
}

this.topologySpecMap = Maps.newConcurrentMap();
this.config = config;
Expand Down Expand Up @@ -197,10 +202,10 @@ private AddSpecResponse onAddFlowSpec(FlowSpec flowSpec) {
response = dag.toString();
}

if (FlowCatalog.isCompileSuccessful(response) && this.warmStandbyEnabled && !flowSpec.isExplain() &&
if (FlowCatalog.isCompileSuccessful(response) && this.userQuotaManager.isPresent() && !flowSpec.isExplain() &&
(!flowSpec.getConfigAsProperties().containsKey(ConfigurationKeys.JOB_SCHEDULE_KEY) || PropertiesUtils.getPropAsBoolean(flowSpec.getConfigAsProperties(), ConfigurationKeys.FLOW_RUN_IMMEDIATELY, "false"))) {
try {
userQuotaManager.checkQuota(dag.getStartNodes());
userQuotaManager.get().checkQuota(dag.getStartNodes());
flowSpec.getConfigAsProperties().setProperty(ServiceConfigKeys.GOBBLIN_SERVICE_ADHOC_FLOW, "true");
} catch (IOException e) {
throw new RuntimeException(e);
Expand Down
3 changes: 2 additions & 1 deletion ...in-service/src/main/java/org/apache/gobblin/service/modules/orchestration/DagManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -390,7 +390,8 @@ public synchronized void setActive(boolean active) {

this.dagManagerMetrics.activate();

UserQuotaManager quotaManager = new InMemoryUserQuotaManager(config);
UserQuotaManager quotaManager = GobblinConstructorUtils.invokeConstructor(UserQuotaManager.class,
ConfigUtils.getString(config, ServiceConfigKeys.QUOTA_MANAGER_CLASS, ServiceConfigKeys.DEFAULT_QUOTA_MANAGER), config);
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

do you see other types of quota manager besides in memory actually being used?

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

In warm standby mode, we need use mysql based quota manager

quotaManager.init(dagStateStore.getDags());

//On startup, the service creates DagManagerThreads that are scheduled at a fixed rate.
Expand Down
14 changes: 11 additions & 3 deletions ...src/main/java/org/apache/gobblin/service/modules/orchestration/MysqlUserQuotaManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -50,15 +50,22 @@
@Slf4j
@Singleton
public class MysqlUserQuotaManager extends AbstractUserQuotaManager {
public final static String CONFIG_PREFIX= "MysqlUserQuotaManager";
public final MysqlQuotaStore quotaStore;
public final RunningDagIdsStore runningDagIds;


@Inject
public MysqlUserQuotaManager(Config config) throws IOException {
super(config);
this.quotaStore = createQuotaStore(config);
this.runningDagIds = createRunningDagStore(config);
Config quotaStoreConfig;
if (config.hasPath(CONFIG_PREFIX)) {
quotaStoreConfig = config.getConfig(CONFIG_PREFIX).withFallback(config);
} else {
throw new IOException("Please specify the config for MysqlUserQuotaManager");
}
this.quotaStore = createQuotaStore(quotaStoreConfig);
this.runningDagIds = createRunningDagStore(quotaStoreConfig);
}

void addDagId(Connection connection, String dagId) throws IOException {
Expand Down Expand Up @@ -295,7 +302,8 @@ public MysqlQuotaStore(BasicDataSource dataSource, String tableName)
DECREASE_FLOWGROUP_COUNT_SQL = "UPDATE " + tableName + " SET flowgroup_count=flowgroup_count-1 WHERE name = ?";
DELETE_USER_SQL = "DELETE FROM " + tableName + " WHERE name = ? AND user_count<1 AND flowgroup_count<1";

String createQuotaTable = "CREATE TABLE IF NOT EXISTS " + tableName + " (name VARCHAR(20) CHARACTER SET latin1 NOT NULL, "
//Increase the length of name as we include the executor uri in it
String createQuotaTable = "CREATE TABLE IF NOT EXISTS " + tableName + " (name VARCHAR(500) CHARACTER SET latin1 NOT NULL, "
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

anything else contained in table name? how long is executor uri max? It will be helpful to give idea of what table name will be

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Under name, it will be "user name for the quota" + "spec uri". I tried 200 and see it does not work so directly give 500

+ "user_count INT NOT NULL DEFAULT 0, requester_count INT NOT NULL DEFAULT 0, flowgroup_count INT NOT NULL DEFAULT 0, "
+ "PRIMARY KEY (name), " + "UNIQUE INDEX ind (name))";
try (Connection connection = dataSource.getConnection(); PreparedStatement createStatement = connection.prepareStatement(createQuotaTable)) {
Expand Down
8 changes: 4 additions & 4 deletions ...test/java/org/apache/gobblin/service/modules/orchestration/MysqlUserQuotaManagerTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -45,10 +45,10 @@ public void setUp() throws Exception {
ITestMetastoreDatabase testDb = TestMetastoreDatabaseFactory.get();

Config config = ConfigBuilder.create()
.addPrimitive(ConfigurationKeys.STATE_STORE_DB_URL_KEY, testDb.getJdbcUrl())
.addPrimitive(ConfigurationKeys.STATE_STORE_DB_USER_KEY, USER)
.addPrimitive(ConfigurationKeys.STATE_STORE_DB_PASSWORD_KEY, PASSWORD)
.addPrimitive(ConfigurationKeys.STATE_STORE_DB_TABLE_KEY, TABLE)
.addPrimitive(MysqlUserQuotaManager.CONFIG_PREFIX + '.' + ConfigurationKeys.STATE_STORE_DB_URL_KEY, testDb.getJdbcUrl())
.addPrimitive(MysqlUserQuotaManager.CONFIG_PREFIX + '.' + ConfigurationKeys.STATE_STORE_DB_USER_KEY, USER)
.addPrimitive(MysqlUserQuotaManager.CONFIG_PREFIX + '.' + ConfigurationKeys.STATE_STORE_DB_PASSWORD_KEY, PASSWORD)
.addPrimitive(MysqlUserQuotaManager.CONFIG_PREFIX + '.' + ConfigurationKeys.STATE_STORE_DB_TABLE_KEY, TABLE)
.build();

this.quotaManager = new MysqlUserQuotaManager(config);
Expand Down