goiri commented on code in PR #5810:
URL: https://github.com/apache/hadoop/pull/5810#discussion_r1265947134
##########
hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/resources/yarn-default.xml:
##########
@@ -5426,4 +5426,36 @@
<value>100ms</value>
</property>
+ <property>
+ <description>
+ The interval at which the policy generator runs, default is one hour
+ </description>
+ <name>yarn.federation.gpg.policy.generator.interval-ms</name>
Review Comment:
Not sure how we handle this with backports but as the metric is general
durations with units, it shouldn't include ms.
##########
hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-globalpolicygenerator/src/main/java/org/apache/hadoop/yarn/server/globalpolicygenerator/GPGPolicyFacade.java:
##########
@@ -0,0 +1,222 @@
+/**
+ * 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.hadoop.yarn.server.globalpolicygenerator;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.yarn.conf.YarnConfiguration;
+import org.apache.hadoop.yarn.exceptions.YarnException;
+import org.apache.hadoop.yarn.server.federation.policies.FederationPolicyUtils;
+import
org.apache.hadoop.yarn.server.federation.policies.dao.WeightedPolicyInfo;
+import
org.apache.hadoop.yarn.server.federation.policies.manager.WeightedLocalityPolicyManager;
+import
org.apache.hadoop.yarn.server.federation.policies.router.FederationRouterPolicy;
+import
org.apache.hadoop.yarn.server.federation.policies.amrmproxy.FederationAMRMProxyPolicy;
+import
org.apache.hadoop.yarn.server.federation.policies.exceptions.FederationPolicyInitializationException;
+import
org.apache.hadoop.yarn.server.federation.policies.manager.FederationPolicyManager;
+import
org.apache.hadoop.yarn.server.federation.store.records.SubClusterPolicyConfiguration;
+import
org.apache.hadoop.yarn.server.federation.utils.FederationStateStoreFacade;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * A utility class for the GPG Policy Generator to read and write policies
+ * into the FederationStateStore. Policy specific logic is abstracted away in
+ * this class, so the PolicyGenerator can avoid dealing with policy
+ * construction, reinitialization, and serialization.
+ *
+ * There are only two exposed methods:
+ *
+ * {@link #getPolicyManager(String)}
+ * Gets the PolicyManager via queue name. Null if there is no policy
+ * configured for the specified queue. The PolicyManager can be used to
+ * extract the {@link FederationRouterPolicy} and
+ * {@link FederationAMRMProxyPolicy}, as well as any policy specific parameters
+ *
+ * {@link #setPolicyManager(FederationPolicyManager)}
+ * Sets the PolicyManager. If the policy configuration is the same, no change
+ * occurs. Otherwise, the internal cache is updated and the new configuration
+ * is written into the FederationStateStore
+ *
+ * This class assumes that the GPG is the only service
+ * writing policies. Thus, the only FederationStateStore reads occur the first
+ * time a queue policy is retrieved - after that, the GPG only writes to the
+ * FederationStateStore.
+ *
+ * The class uses a PolicyManager cache and a SubClusterPolicyConfiguration
+ * cache. The primary use for these caches are to serve reads, and to
+ * identify when the PolicyGenerator has actually changed the policy
+ * so unnecessary FederationStateStore policy writes can be avoided.
+ */
+
+public class GPGPolicyFacade {
+
+ private static final Logger LOG =
+ LoggerFactory.getLogger(GPGPolicyFacade.class);
+
+ private FederationStateStoreFacade stateStore;
+
+ private Map<String, FederationPolicyManager> policyManagerMap;
+ private Map<String, SubClusterPolicyConfiguration> policyConfMap;
+
+ private boolean readOnly;
+
+ public GPGPolicyFacade(FederationStateStoreFacade stateStore,
+ Configuration conf) {
+ this.stateStore = stateStore;
+ this.policyManagerMap = new HashMap<>();
+ this.policyConfMap = new HashMap<>();
+ this.readOnly =
+ conf.getBoolean(YarnConfiguration.GPG_POLICY_GENERATOR_READONLY,
+ YarnConfiguration.DEFAULT_GPG_POLICY_GENERATOR_READONLY);
+ }
+
+ /**
+ * Provides a utility for the policy generator to read the policy manager
+ * from the FederationStateStore. Because the policy generator should be the
+ * only component updating the policy, this implementation does not use the
+ * reinitialization feature.
+ *
+ * @param queueName the name of the queue we want the policy manager for.
+ * @return the policy manager responsible for the queue policy.
+ * @throws YarnException exceptions from yarn servers.
+ */
+ public FederationPolicyManager getPolicyManager(String queueName)
+ throws YarnException {
+ FederationPolicyManager policyManager = policyManagerMap.get(queueName);
+ // If we don't have the policy manager cached, pull configuration
+ // from the FederationStateStore to create and cache it
+ if (policyManager == null) {
+ try {
+ // If we don't have the configuration cached, pull it
+ // from the stateStore
+ SubClusterPolicyConfiguration conf = policyConfMap.get(queueName);
+ if (conf == null) {
+ conf = stateStore.getPolicyConfiguration(queueName);
+ }
+ // If configuration is still null, it does not exist in the
+ // FederationStateStore
+ if (conf == null) {
+ LOG.info("Read null policy for queue {}", queueName);
+ return null;
+ }
+ policyManager =
+ FederationPolicyUtils.instantiatePolicyManager(conf.getType());
+ policyManager.setQueue(queueName);
+
+ // TODO there is currently no way to cleanly deserialize a policy
Review Comment:
Are we planning on fixing this?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: common-issues-unsubscr...@hadoop.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: common-issues-unsubscr...@hadoop.apache.org
For additional commands, e-mail: common-issues-h...@hadoop.apache.org
