YARN-11293. [Federation] StoreNewMasterKey/removeStoredMasterKey With MemoryStateStore. (#4852)
This commit is contained in:
parent
620dd37712
commit
342c4856b8
@ -196,6 +196,9 @@
|
||||
<additionalProtoPathElement>
|
||||
${basedir}/../../hadoop-yarn-api/src/main/proto
|
||||
</additionalProtoPathElement>
|
||||
<additionalProtoPathElement>
|
||||
${basedir}/../../hadoop-yarn-common/src/main/proto
|
||||
</additionalProtoPathElement>
|
||||
</additionalProtoPathElements>
|
||||
</configuration>
|
||||
</execution>
|
||||
|
@ -0,0 +1,69 @@
|
||||
/**
|
||||
* 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
|
||||
* <p>
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
* <p>
|
||||
* 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.federation.store;
|
||||
|
||||
import org.apache.hadoop.classification.InterfaceAudience.Private;
|
||||
import org.apache.hadoop.classification.InterfaceStability.Unstable;
|
||||
import org.apache.hadoop.yarn.exceptions.YarnException;
|
||||
import org.apache.hadoop.yarn.server.federation.store.records.RouterMasterKeyRequest;
|
||||
import org.apache.hadoop.yarn.server.federation.store.records.RouterMasterKeyResponse;
|
||||
|
||||
import java.io.IOException;
|
||||
|
||||
/**
|
||||
* FederationDelegationTokenStateStore maintains the state of all
|
||||
* <em>DelegationToken</em> that have been submitted to the federated cluster.
|
||||
*/
|
||||
@Private
|
||||
@Unstable
|
||||
public interface FederationDelegationTokenStateStore {
|
||||
|
||||
/**
|
||||
* The Router Supports Store NewMasterKey.
|
||||
* During this Process, Facade will call the specific StateStore to store the MasterKey.
|
||||
*
|
||||
* @param request The request contains RouterMasterKey, which is an abstraction for DelegationKey
|
||||
* @return routerMasterKeyResponse
|
||||
* @throws YarnException if the call to the state store is unsuccessful
|
||||
* @throws IOException An IO Error occurred
|
||||
*/
|
||||
RouterMasterKeyResponse storeNewMasterKey(RouterMasterKeyRequest request)
|
||||
throws YarnException, IOException;
|
||||
|
||||
/**
|
||||
* The Router Supports Remove MasterKey.
|
||||
* During this Process, Facade will call the specific StateStore to remove the MasterKey.
|
||||
*
|
||||
* @param request The request contains RouterMasterKey, which is an abstraction for DelegationKey
|
||||
* @return routerMasterKeyResponse
|
||||
* @throws YarnException if the call to the state store is unsuccessful
|
||||
* @throws IOException An IO Error occurred
|
||||
*/
|
||||
RouterMasterKeyResponse removeStoredMasterKey(RouterMasterKeyRequest request)
|
||||
throws YarnException, IOException;
|
||||
|
||||
/**
|
||||
* The Router Supports GetMasterKeyByDelegationKey.
|
||||
*
|
||||
* @param request The request contains RouterMasterKey, which is an abstraction for DelegationKey
|
||||
* @return routerMasterKeyResponse
|
||||
* @throws YarnException if the call to the state store is unsuccessful
|
||||
* @throws IOException An IO Error occurred
|
||||
*/
|
||||
RouterMasterKeyResponse getMasterKeyByDelegationKey(RouterMasterKeyRequest request)
|
||||
throws YarnException, IOException;
|
||||
}
|
@ -31,7 +31,8 @@
|
||||
*/
|
||||
public interface FederationStateStore extends
|
||||
FederationApplicationHomeSubClusterStore, FederationMembershipStateStore,
|
||||
FederationPolicyStore, FederationReservationHomeSubClusterStore {
|
||||
FederationPolicyStore, FederationReservationHomeSubClusterStore,
|
||||
FederationDelegationTokenStateStore {
|
||||
|
||||
/**
|
||||
* Initialize the FederationStore.
|
||||
|
@ -17,15 +17,20 @@
|
||||
|
||||
package org.apache.hadoop.yarn.server.federation.store.impl;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.nio.ByteBuffer;
|
||||
import java.util.ArrayList;
|
||||
import java.util.Calendar;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.Map.Entry;
|
||||
import java.util.Set;
|
||||
import java.util.TimeZone;
|
||||
import java.util.concurrent.ConcurrentHashMap;
|
||||
|
||||
import org.apache.hadoop.classification.VisibleForTesting;
|
||||
import org.apache.hadoop.conf.Configuration;
|
||||
import org.apache.hadoop.security.token.delegation.DelegationKey;
|
||||
import org.apache.hadoop.yarn.api.records.ApplicationId;
|
||||
import org.apache.hadoop.yarn.api.records.ReservationId;
|
||||
import org.apache.hadoop.yarn.exceptions.YarnException;
|
||||
@ -71,6 +76,10 @@
|
||||
import org.apache.hadoop.yarn.server.federation.store.records.UpdateReservationHomeSubClusterResponse;
|
||||
import org.apache.hadoop.yarn.server.federation.store.records.DeleteReservationHomeSubClusterRequest;
|
||||
import org.apache.hadoop.yarn.server.federation.store.records.DeleteReservationHomeSubClusterResponse;
|
||||
import org.apache.hadoop.yarn.server.federation.store.records.RouterMasterKey;
|
||||
import org.apache.hadoop.yarn.server.federation.store.records.RouterMasterKeyRequest;
|
||||
import org.apache.hadoop.yarn.server.federation.store.records.RouterMasterKeyResponse;
|
||||
import org.apache.hadoop.yarn.server.federation.store.records.RouterRMDTSecretManagerState;
|
||||
import org.apache.hadoop.yarn.server.federation.store.utils.FederationApplicationHomeSubClusterStoreInputValidator;
|
||||
import org.apache.hadoop.yarn.server.federation.store.utils.FederationReservationHomeSubClusterStoreInputValidator;
|
||||
import org.apache.hadoop.yarn.server.federation.store.utils.FederationMembershipStateStoreInputValidator;
|
||||
@ -90,6 +99,7 @@ public class MemoryFederationStateStore implements FederationStateStore {
|
||||
private Map<ApplicationId, SubClusterId> applications;
|
||||
private Map<ReservationId, SubClusterId> reservations;
|
||||
private Map<String, SubClusterPolicyConfiguration> policies;
|
||||
private RouterRMDTSecretManagerState routerRMSecretManagerState;
|
||||
|
||||
private final MonotonicClock clock = new MonotonicClock();
|
||||
|
||||
@ -102,6 +112,7 @@ public void init(Configuration conf) {
|
||||
applications = new ConcurrentHashMap<ApplicationId, SubClusterId>();
|
||||
reservations = new ConcurrentHashMap<ReservationId, SubClusterId>();
|
||||
policies = new ConcurrentHashMap<String, SubClusterPolicyConfiguration>();
|
||||
routerRMSecretManagerState = new RouterRMDTSecretManagerState();
|
||||
}
|
||||
|
||||
@Override
|
||||
@ -395,4 +406,74 @@ public DeleteReservationHomeSubClusterResponse deleteReservationHomeSubCluster(
|
||||
reservations.remove(reservationId);
|
||||
return DeleteReservationHomeSubClusterResponse.newInstance();
|
||||
}
|
||||
|
||||
@Override
|
||||
public RouterMasterKeyResponse storeNewMasterKey(RouterMasterKeyRequest request)
|
||||
throws YarnException, IOException {
|
||||
// Restore the DelegationKey from the request
|
||||
RouterMasterKey masterKey = request.getRouterMasterKey();
|
||||
DelegationKey delegationKey = getDelegationKeyByMasterKey(masterKey);
|
||||
|
||||
Set<DelegationKey> rmDTMasterKeyState = routerRMSecretManagerState.getMasterKeyState();
|
||||
if (rmDTMasterKeyState.contains(delegationKey)) {
|
||||
LOG.info("Error storing info for RMDTMasterKey with keyID: {}.", delegationKey.getKeyId());
|
||||
throw new IOException("RMDTMasterKey with keyID: " + delegationKey.getKeyId() +
|
||||
" is already stored");
|
||||
}
|
||||
|
||||
routerRMSecretManagerState.getMasterKeyState().add(delegationKey);
|
||||
LOG.info("Store Router-RMDT master key with key id: {}. Currently rmDTMasterKeyState size: {}",
|
||||
delegationKey.getKeyId(), rmDTMasterKeyState.size());
|
||||
|
||||
return RouterMasterKeyResponse.newInstance(masterKey);
|
||||
}
|
||||
|
||||
@Override
|
||||
public RouterMasterKeyResponse removeStoredMasterKey(RouterMasterKeyRequest request)
|
||||
throws YarnException, IOException {
|
||||
// Restore the DelegationKey from the request
|
||||
RouterMasterKey masterKey = request.getRouterMasterKey();
|
||||
DelegationKey delegationKey = getDelegationKeyByMasterKey(masterKey);
|
||||
|
||||
LOG.info("Remove Router-RMDT master key with key id: {}.", delegationKey.getKeyId());
|
||||
Set<DelegationKey> rmDTMasterKeyState = routerRMSecretManagerState.getMasterKeyState();
|
||||
rmDTMasterKeyState.remove(delegationKey);
|
||||
|
||||
return RouterMasterKeyResponse.newInstance(masterKey);
|
||||
}
|
||||
|
||||
@Override
|
||||
public RouterMasterKeyResponse getMasterKeyByDelegationKey(RouterMasterKeyRequest request)
|
||||
throws YarnException, IOException {
|
||||
// Restore the DelegationKey from the request
|
||||
RouterMasterKey masterKey = request.getRouterMasterKey();
|
||||
DelegationKey delegationKey = getDelegationKeyByMasterKey(masterKey);
|
||||
|
||||
Set<DelegationKey> rmDTMasterKeyState = routerRMSecretManagerState.getMasterKeyState();
|
||||
if (!rmDTMasterKeyState.contains(delegationKey)) {
|
||||
throw new IOException("GetMasterKey with keyID: " + masterKey.getKeyId() +
|
||||
" does not exist.");
|
||||
}
|
||||
RouterMasterKey resultRouterMasterKey = RouterMasterKey.newInstance(delegationKey.getKeyId(),
|
||||
ByteBuffer.wrap(delegationKey.getEncodedKey()), delegationKey.getExpiryDate());
|
||||
return RouterMasterKeyResponse.newInstance(resultRouterMasterKey);
|
||||
}
|
||||
|
||||
/**
|
||||
* Get DelegationKey By based on MasterKey.
|
||||
*
|
||||
* @param masterKey masterKey
|
||||
* @return DelegationKey
|
||||
*/
|
||||
private static DelegationKey getDelegationKeyByMasterKey(RouterMasterKey masterKey) {
|
||||
ByteBuffer keyByteBuf = masterKey.getKeyBytes();
|
||||
byte[] keyBytes = new byte[keyByteBuf.remaining()];
|
||||
keyByteBuf.get(keyBytes);
|
||||
return new DelegationKey(masterKey.getKeyId(), masterKey.getExpiryDate(), keyBytes);
|
||||
}
|
||||
|
||||
@VisibleForTesting
|
||||
public RouterRMDTSecretManagerState getRouterRMSecretManagerState() {
|
||||
return routerRMSecretManagerState;
|
||||
}
|
||||
}
|
||||
|
@ -18,6 +18,7 @@
|
||||
|
||||
package org.apache.hadoop.yarn.server.federation.store.impl;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.nio.ByteBuffer;
|
||||
import java.sql.CallableStatement;
|
||||
import java.sql.Connection;
|
||||
@ -81,6 +82,8 @@
|
||||
import org.apache.hadoop.yarn.server.federation.store.records.UpdateReservationHomeSubClusterRequest;
|
||||
import org.apache.hadoop.yarn.server.federation.store.records.UpdateReservationHomeSubClusterResponse;
|
||||
import org.apache.hadoop.yarn.server.federation.store.records.ReservationHomeSubCluster;
|
||||
import org.apache.hadoop.yarn.server.federation.store.records.RouterMasterKeyRequest;
|
||||
import org.apache.hadoop.yarn.server.federation.store.records.RouterMasterKeyResponse;
|
||||
import org.apache.hadoop.yarn.server.federation.store.utils.FederationApplicationHomeSubClusterStoreInputValidator;
|
||||
import org.apache.hadoop.yarn.server.federation.store.utils.FederationMembershipStateStoreInputValidator;
|
||||
import org.apache.hadoop.yarn.server.federation.store.utils.FederationPolicyStoreInputValidator;
|
||||
@ -1385,4 +1388,22 @@ public UpdateReservationHomeSubClusterResponse updateReservationHomeSubCluster(
|
||||
"Unable to update the subCluster " + subClusterId +
|
||||
" according to reservation" + reservationId);
|
||||
}
|
||||
|
||||
@Override
|
||||
public RouterMasterKeyResponse storeNewMasterKey(RouterMasterKeyRequest request)
|
||||
throws YarnException, IOException {
|
||||
throw new NotImplementedException("Code is not implemented");
|
||||
}
|
||||
|
||||
@Override
|
||||
public RouterMasterKeyResponse removeStoredMasterKey(RouterMasterKeyRequest request)
|
||||
throws YarnException, IOException {
|
||||
throw new NotImplementedException("Code is not implemented");
|
||||
}
|
||||
|
||||
@Override
|
||||
public RouterMasterKeyResponse getMasterKeyByDelegationKey(RouterMasterKeyRequest request)
|
||||
throws YarnException, IOException {
|
||||
throw new NotImplementedException("Code is not implemented");
|
||||
}
|
||||
}
|
||||
|
@ -25,6 +25,7 @@
|
||||
import java.util.List;
|
||||
import java.util.TimeZone;
|
||||
|
||||
import org.apache.commons.lang3.NotImplementedException;
|
||||
import org.apache.hadoop.classification.VisibleForTesting;
|
||||
import org.apache.hadoop.conf.Configuration;
|
||||
import org.apache.hadoop.util.curator.ZKCuratorManager;
|
||||
@ -77,6 +78,8 @@
|
||||
import org.apache.hadoop.yarn.server.federation.store.records.DeleteReservationHomeSubClusterResponse;
|
||||
import org.apache.hadoop.yarn.server.federation.store.records.UpdateReservationHomeSubClusterRequest;
|
||||
import org.apache.hadoop.yarn.server.federation.store.records.UpdateReservationHomeSubClusterResponse;
|
||||
import org.apache.hadoop.yarn.server.federation.store.records.RouterMasterKeyResponse;
|
||||
import org.apache.hadoop.yarn.server.federation.store.records.RouterMasterKeyRequest;
|
||||
import org.apache.hadoop.yarn.server.federation.store.records.impl.pb.SubClusterIdPBImpl;
|
||||
import org.apache.hadoop.yarn.server.federation.store.records.impl.pb.SubClusterInfoPBImpl;
|
||||
import org.apache.hadoop.yarn.server.federation.store.records.impl.pb.SubClusterPolicyConfigurationPBImpl;
|
||||
@ -851,4 +854,22 @@ public UpdateReservationHomeSubClusterResponse updateReservationHomeSubCluster(
|
||||
opDurations.addUpdateReservationHomeSubClusterDuration(start, end);
|
||||
return UpdateReservationHomeSubClusterResponse.newInstance();
|
||||
}
|
||||
|
||||
@Override
|
||||
public RouterMasterKeyResponse storeNewMasterKey(RouterMasterKeyRequest request)
|
||||
throws YarnException, IOException {
|
||||
throw new NotImplementedException("Code is not implemented");
|
||||
}
|
||||
|
||||
@Override
|
||||
public RouterMasterKeyResponse removeStoredMasterKey(RouterMasterKeyRequest request)
|
||||
throws YarnException, IOException {
|
||||
throw new NotImplementedException("Code is not implemented");
|
||||
}
|
||||
|
||||
@Override
|
||||
public RouterMasterKeyResponse getMasterKeyByDelegationKey(RouterMasterKeyRequest request)
|
||||
throws YarnException, IOException {
|
||||
throw new NotImplementedException("Code is not implemented");
|
||||
}
|
||||
}
|
@ -0,0 +1,133 @@
|
||||
/**
|
||||
* 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.federation.store.records;
|
||||
|
||||
import org.apache.commons.lang3.builder.EqualsBuilder;
|
||||
import org.apache.commons.lang3.builder.HashCodeBuilder;
|
||||
import org.apache.hadoop.classification.InterfaceAudience.Private;
|
||||
import org.apache.hadoop.classification.InterfaceAudience.Public;
|
||||
import org.apache.hadoop.classification.InterfaceStability.Unstable;
|
||||
import org.apache.hadoop.yarn.util.Records;
|
||||
|
||||
import java.nio.ByteBuffer;
|
||||
|
||||
@Private
|
||||
@Unstable
|
||||
public abstract class RouterMasterKey {
|
||||
|
||||
@Private
|
||||
@Unstable
|
||||
public static RouterMasterKey newInstance(Integer keyId, ByteBuffer keyBytes, Long expiryDate) {
|
||||
RouterMasterKey policy = Records.newRecord(RouterMasterKey.class);
|
||||
policy.setKeyId(keyId);
|
||||
policy.setKeyBytes(keyBytes);
|
||||
policy.setExpiryDate(expiryDate);
|
||||
return policy;
|
||||
}
|
||||
|
||||
@Private
|
||||
@Unstable
|
||||
public static RouterMasterKey newInstance(RouterMasterKey masterKey) {
|
||||
RouterMasterKey routerMasterKey = Records.newRecord(RouterMasterKey.class);
|
||||
routerMasterKey.setKeyId(masterKey.getKeyId());
|
||||
routerMasterKey.setKeyBytes(masterKey.getKeyBytes());
|
||||
routerMasterKey.setExpiryDate(masterKey.getExpiryDate());
|
||||
return routerMasterKey;
|
||||
}
|
||||
|
||||
/**
|
||||
* Get the keyId of the MasterKey.
|
||||
*
|
||||
* @return MasterKeyId.
|
||||
*/
|
||||
@Public
|
||||
@Unstable
|
||||
public abstract Integer getKeyId();
|
||||
|
||||
/**
|
||||
* Set the keyId of the MasterKey.
|
||||
*
|
||||
* @param keyId MasterKeyId.
|
||||
*/
|
||||
@Private
|
||||
@Unstable
|
||||
public abstract void setKeyId(Integer keyId);
|
||||
|
||||
/**
|
||||
* Get the keyBytes of the DelegationKey.
|
||||
*
|
||||
* @return KeyBytes of the DelegationKey.
|
||||
*/
|
||||
@Public
|
||||
@Unstable
|
||||
public abstract ByteBuffer getKeyBytes();
|
||||
|
||||
/**
|
||||
* Set the keyBytes of the DelegationKey.
|
||||
*
|
||||
* @param keyBytes KeyBytes of the DelegationKey.
|
||||
*/
|
||||
@Private
|
||||
@Unstable
|
||||
public abstract void setKeyBytes(ByteBuffer keyBytes);
|
||||
|
||||
/**
|
||||
* Get the ExpiryDate of the DelegationKey.
|
||||
*
|
||||
* @return ExpiryDate of the DelegationKey.
|
||||
*/
|
||||
@Private
|
||||
@Unstable
|
||||
public abstract Long getExpiryDate();
|
||||
|
||||
/**
|
||||
* Set the expiryDate of the DelegationKey.
|
||||
*
|
||||
* @param expiryDate expiryDate of the DelegationKey.
|
||||
*/
|
||||
@Private
|
||||
@Unstable
|
||||
public abstract void setExpiryDate(Long expiryDate);
|
||||
|
||||
@Override
|
||||
public int hashCode() {
|
||||
return new HashCodeBuilder()
|
||||
.append(this.getExpiryDate().longValue())
|
||||
.append(this.getKeyId().intValue())
|
||||
.append(getKeyBytes().array())
|
||||
.hashCode();
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean equals(Object right) {
|
||||
if (this == right) {
|
||||
return true;
|
||||
}
|
||||
|
||||
if (right == null || getClass() != right.getClass()) {
|
||||
return false;
|
||||
}
|
||||
|
||||
RouterMasterKey r = (RouterMasterKey) right;
|
||||
return new EqualsBuilder()
|
||||
.append(this.getKeyId().intValue(), r.getKeyId().intValue())
|
||||
.append(this.getExpiryDate().longValue(), this.getExpiryDate().longValue())
|
||||
.append(getKeyBytes().array(), r.getKeyBytes())
|
||||
.isEquals();
|
||||
}
|
||||
}
|
@ -0,0 +1,43 @@
|
||||
/**
|
||||
* 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
|
||||
* <p>
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
* <p>
|
||||
* 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.federation.store.records;
|
||||
|
||||
import org.apache.hadoop.classification.InterfaceAudience.Private;
|
||||
import org.apache.hadoop.classification.InterfaceAudience.Public;
|
||||
import org.apache.hadoop.classification.InterfaceStability.Unstable;
|
||||
import org.apache.hadoop.yarn.util.Records;
|
||||
|
||||
@Private
|
||||
@Unstable
|
||||
public abstract class RouterMasterKeyRequest {
|
||||
|
||||
@Private
|
||||
@Unstable
|
||||
public static RouterMasterKeyRequest newInstance(RouterMasterKey routerMasterKey) {
|
||||
RouterMasterKeyRequest request = Records.newRecord(RouterMasterKeyRequest.class);
|
||||
request.setRouterMasterKey(routerMasterKey);
|
||||
return request;
|
||||
}
|
||||
|
||||
@Public
|
||||
@Unstable
|
||||
public abstract RouterMasterKey getRouterMasterKey();
|
||||
|
||||
@Private
|
||||
@Unstable
|
||||
public abstract void setRouterMasterKey(RouterMasterKey routerMasterKey);
|
||||
}
|
@ -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.hadoop.yarn.server.federation.store.records;
|
||||
|
||||
import org.apache.hadoop.classification.InterfaceAudience.Private;
|
||||
import org.apache.hadoop.classification.InterfaceAudience.Public;
|
||||
import org.apache.hadoop.classification.InterfaceStability.Unstable;
|
||||
import org.apache.hadoop.yarn.util.Records;
|
||||
|
||||
@Private
|
||||
@Unstable
|
||||
public abstract class RouterMasterKeyResponse {
|
||||
|
||||
@Private
|
||||
@Unstable
|
||||
public static RouterMasterKeyResponse newInstance(RouterMasterKey masterKey) {
|
||||
RouterMasterKeyResponse request = Records.newRecord(RouterMasterKeyResponse.class);
|
||||
request.setRouterMasterKey(masterKey);
|
||||
return request;
|
||||
}
|
||||
|
||||
@Public
|
||||
@Unstable
|
||||
public abstract RouterMasterKey getRouterMasterKey();
|
||||
|
||||
@Private
|
||||
@Unstable
|
||||
public abstract void setRouterMasterKey(RouterMasterKey masterKey);
|
||||
}
|
@ -0,0 +1,52 @@
|
||||
/**
|
||||
* 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.federation.store.records;
|
||||
|
||||
import org.apache.hadoop.security.token.delegation.DelegationKey;
|
||||
import org.apache.hadoop.yarn.security.client.RMDelegationTokenIdentifier;
|
||||
|
||||
import java.util.HashMap;
|
||||
import java.util.HashSet;
|
||||
import java.util.Map;
|
||||
import java.util.Set;
|
||||
|
||||
public class RouterRMDTSecretManagerState {
|
||||
|
||||
// DTIdentifier -> renewDate
|
||||
private Map<RMDelegationTokenIdentifier, Long> delegationTokenState = new HashMap<>();
|
||||
|
||||
private Set<DelegationKey> masterKeyState = new HashSet<>();
|
||||
|
||||
private int dtSequenceNumber = 0;
|
||||
|
||||
public Map<RMDelegationTokenIdentifier, Long> getTokenState() {
|
||||
return delegationTokenState;
|
||||
}
|
||||
|
||||
public Set<DelegationKey> getMasterKeyState() {
|
||||
return masterKeyState;
|
||||
}
|
||||
|
||||
public int getDTSequenceNumber() {
|
||||
return dtSequenceNumber;
|
||||
}
|
||||
|
||||
public void setDtSequenceNumber(int dtSequenceNumber) {
|
||||
this.dtSequenceNumber = dtSequenceNumber;
|
||||
}
|
||||
}
|
@ -0,0 +1,133 @@
|
||||
/**
|
||||
* 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.federation.store.records.impl.pb;
|
||||
|
||||
import org.apache.hadoop.thirdparty.protobuf.ByteString;
|
||||
import org.apache.hadoop.thirdparty.protobuf.TextFormat;
|
||||
import org.apache.hadoop.yarn.api.records.impl.pb.ProtoUtils;
|
||||
import org.apache.hadoop.yarn.federation.proto.YarnServerFederationProtos.RouterMasterKeyProtoOrBuilder;
|
||||
import org.apache.hadoop.yarn.federation.proto.YarnServerFederationProtos.RouterMasterKeyProto;
|
||||
import org.apache.hadoop.yarn.server.federation.store.records.RouterMasterKey;
|
||||
|
||||
import java.nio.ByteBuffer;
|
||||
|
||||
public class RouterMasterKeyPBImpl extends RouterMasterKey {
|
||||
|
||||
private RouterMasterKeyProto proto = RouterMasterKeyProto.getDefaultInstance();
|
||||
private RouterMasterKeyProto.Builder builder = null;
|
||||
private boolean viaProto = false;
|
||||
|
||||
public RouterMasterKeyPBImpl() {
|
||||
builder = RouterMasterKeyProto.newBuilder();
|
||||
}
|
||||
|
||||
public RouterMasterKeyPBImpl(RouterMasterKeyProto masterKeyProto) {
|
||||
this.proto = masterKeyProto;
|
||||
viaProto = true;
|
||||
}
|
||||
|
||||
public RouterMasterKeyProto getProto() {
|
||||
proto = viaProto ? proto : builder.build();
|
||||
viaProto = true;
|
||||
return proto;
|
||||
}
|
||||
|
||||
private void maybeInitBuilder() {
|
||||
if (viaProto || builder == null) {
|
||||
builder = RouterMasterKeyProto.newBuilder(proto);
|
||||
}
|
||||
viaProto = false;
|
||||
}
|
||||
|
||||
@Override
|
||||
public int hashCode() {
|
||||
return getProto().hashCode();
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean equals(Object other) {
|
||||
if (other == null) {
|
||||
return false;
|
||||
}
|
||||
if (other.getClass().isAssignableFrom(this.getClass())) {
|
||||
return this.getProto().equals(this.getClass().cast(other).getProto());
|
||||
}
|
||||
return false;
|
||||
}
|
||||
|
||||
@Override
|
||||
public String toString() {
|
||||
return TextFormat.shortDebugString(getProto());
|
||||
}
|
||||
|
||||
@Override
|
||||
public Integer getKeyId() {
|
||||
RouterMasterKeyProtoOrBuilder p = viaProto ? proto : builder;
|
||||
return p.getKeyId();
|
||||
}
|
||||
|
||||
@Override
|
||||
public void setKeyId(Integer keyId) {
|
||||
maybeInitBuilder();
|
||||
if (keyId == null) {
|
||||
builder.clearKeyId();
|
||||
return;
|
||||
}
|
||||
builder.setKeyId(keyId);
|
||||
}
|
||||
|
||||
@Override
|
||||
public ByteBuffer getKeyBytes() {
|
||||
RouterMasterKeyProtoOrBuilder p = viaProto ? proto : builder;
|
||||
return convertFromProtoFormat(p.getKeyBytes());
|
||||
}
|
||||
|
||||
@Override
|
||||
public void setKeyBytes(ByteBuffer keyBytes) {
|
||||
maybeInitBuilder();
|
||||
if (keyBytes == null) {
|
||||
builder.clearKeyBytes();
|
||||
return;
|
||||
}
|
||||
builder.setKeyBytes(convertToProtoFormat(keyBytes));
|
||||
}
|
||||
|
||||
@Override
|
||||
public Long getExpiryDate() {
|
||||
RouterMasterKeyProtoOrBuilder p = viaProto ? proto : builder;
|
||||
return p.getExpiryDate();
|
||||
}
|
||||
|
||||
@Override
|
||||
public void setExpiryDate(Long expiryDate) {
|
||||
maybeInitBuilder();
|
||||
if (expiryDate == null) {
|
||||
builder.clearExpiryDate();
|
||||
return;
|
||||
}
|
||||
builder.setExpiryDate(expiryDate);
|
||||
}
|
||||
|
||||
protected final ByteBuffer convertFromProtoFormat(ByteString byteString) {
|
||||
return ProtoUtils.convertFromProtoFormat(byteString);
|
||||
}
|
||||
|
||||
protected final ByteString convertToProtoFormat(ByteBuffer byteBuffer) {
|
||||
return ProtoUtils.convertToProtoFormat(byteBuffer);
|
||||
}
|
||||
}
|
@ -0,0 +1,128 @@
|
||||
/**
|
||||
* 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.federation.store.records.impl.pb;
|
||||
|
||||
import org.apache.hadoop.classification.InterfaceAudience.Private;
|
||||
import org.apache.hadoop.classification.InterfaceStability.Unstable;
|
||||
import org.apache.hadoop.yarn.server.federation.store.records.RouterMasterKey;
|
||||
import org.apache.hadoop.yarn.server.federation.store.records.RouterMasterKeyRequest;
|
||||
import org.apache.hadoop.yarn.federation.proto.YarnServerFederationProtos.RouterMasterKeyProto;
|
||||
import org.apache.hadoop.yarn.federation.proto.YarnServerFederationProtos.RouterMasterKeyRequestProto;
|
||||
import org.apache.hadoop.yarn.federation.proto.YarnServerFederationProtos.RouterMasterKeyRequestProtoOrBuilder;
|
||||
import org.apache.hadoop.thirdparty.protobuf.TextFormat;
|
||||
|
||||
@Private
|
||||
@Unstable
|
||||
public class RouterMasterKeyRequestPBImpl extends RouterMasterKeyRequest {
|
||||
|
||||
private RouterMasterKeyRequestProto proto = RouterMasterKeyRequestProto.getDefaultInstance();
|
||||
private RouterMasterKeyRequestProto.Builder builder = null;
|
||||
private boolean viaProto = false;
|
||||
private RouterMasterKey routerMasterKey = null;
|
||||
|
||||
public RouterMasterKeyRequestPBImpl() {
|
||||
builder = RouterMasterKeyRequestProto.newBuilder();
|
||||
}
|
||||
|
||||
public RouterMasterKeyRequestPBImpl(RouterMasterKeyRequestProto proto) {
|
||||
this.proto = proto;
|
||||
viaProto = true;
|
||||
}
|
||||
|
||||
public RouterMasterKeyRequestProto getProto() {
|
||||
mergeLocalToProto();
|
||||
proto = viaProto ? proto : builder.build();
|
||||
viaProto = true;
|
||||
return proto;
|
||||
}
|
||||
|
||||
private void mergeLocalToProto() {
|
||||
if (viaProto) {
|
||||
maybeInitBuilder();
|
||||
}
|
||||
mergeLocalToBuilder();
|
||||
proto = builder.build();
|
||||
viaProto = true;
|
||||
}
|
||||
|
||||
private void maybeInitBuilder() {
|
||||
if (viaProto || builder == null) {
|
||||
builder = RouterMasterKeyRequestProto.newBuilder(proto);
|
||||
}
|
||||
viaProto = false;
|
||||
}
|
||||
|
||||
private void mergeLocalToBuilder() {
|
||||
RouterMasterKeyPBImpl masterKeyRequest = (RouterMasterKeyPBImpl) this.routerMasterKey;
|
||||
RouterMasterKeyProto routerMasterKeyProto = builder.getRouterMasterKey();
|
||||
if (this.routerMasterKey != null && !masterKeyRequest.getProto().equals(routerMasterKeyProto)) {
|
||||
builder.setRouterMasterKey(convertToProtoFormat(this.routerMasterKey));
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public RouterMasterKey getRouterMasterKey() {
|
||||
RouterMasterKeyRequestProtoOrBuilder p = viaProto ? proto : builder;
|
||||
if (this.routerMasterKey != null) {
|
||||
return this.routerMasterKey;
|
||||
}
|
||||
if (!p.hasRouterMasterKey()) {
|
||||
return null;
|
||||
}
|
||||
this.routerMasterKey = convertFromProtoFormat(p.getRouterMasterKey());
|
||||
return this.routerMasterKey;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void setRouterMasterKey(RouterMasterKey masterKey) {
|
||||
maybeInitBuilder();
|
||||
if (masterKey == null) {
|
||||
builder.clearRouterMasterKey();
|
||||
}
|
||||
this.routerMasterKey = masterKey;
|
||||
}
|
||||
|
||||
@Override
|
||||
public int hashCode() {
|
||||
return getProto().hashCode();
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean equals(Object other) {
|
||||
if (other == null) {
|
||||
return false;
|
||||
}
|
||||
if (other.getClass().isAssignableFrom(this.getClass())) {
|
||||
return this.getProto().equals(this.getClass().cast(other).getProto());
|
||||
}
|
||||
return false;
|
||||
}
|
||||
|
||||
@Override
|
||||
public String toString() {
|
||||
return TextFormat.shortDebugString(getProto());
|
||||
}
|
||||
|
||||
private RouterMasterKey convertFromProtoFormat(RouterMasterKeyProto masterKeyProto) {
|
||||
return new RouterMasterKeyPBImpl(masterKeyProto);
|
||||
}
|
||||
|
||||
private RouterMasterKeyProto convertToProtoFormat(RouterMasterKey masterKey) {
|
||||
return ((RouterMasterKeyPBImpl) masterKey).getProto();
|
||||
}
|
||||
}
|
@ -0,0 +1,128 @@
|
||||
/**
|
||||
* 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.federation.store.records.impl.pb;
|
||||
|
||||
import org.apache.hadoop.classification.InterfaceAudience.Private;
|
||||
import org.apache.hadoop.classification.InterfaceStability.Unstable;
|
||||
import org.apache.hadoop.thirdparty.protobuf.TextFormat;
|
||||
import org.apache.hadoop.yarn.federation.proto.YarnServerFederationProtos.RouterMasterKeyProto;
|
||||
import org.apache.hadoop.yarn.federation.proto.YarnServerFederationProtos.RouterMasterKeyResponseProto;
|
||||
import org.apache.hadoop.yarn.federation.proto.YarnServerFederationProtos.RouterMasterKeyResponseProtoOrBuilder;
|
||||
import org.apache.hadoop.yarn.server.federation.store.records.RouterMasterKey;
|
||||
import org.apache.hadoop.yarn.server.federation.store.records.RouterMasterKeyResponse;
|
||||
|
||||
@Private
|
||||
@Unstable
|
||||
public class RouterMasterKeyResponsePBImpl extends RouterMasterKeyResponse {
|
||||
|
||||
private RouterMasterKeyResponseProto proto = RouterMasterKeyResponseProto.getDefaultInstance();
|
||||
private RouterMasterKeyResponseProto.Builder builder = null;
|
||||
private boolean viaProto = false;
|
||||
private RouterMasterKey routerMasterKey = null;
|
||||
|
||||
public RouterMasterKeyResponsePBImpl() {
|
||||
builder = RouterMasterKeyResponseProto.newBuilder();
|
||||
}
|
||||
|
||||
public RouterMasterKeyResponsePBImpl(RouterMasterKeyResponseProto proto) {
|
||||
this.proto = proto;
|
||||
viaProto = true;
|
||||
}
|
||||
|
||||
public RouterMasterKeyResponseProto getProto() {
|
||||
mergeLocalToProto();
|
||||
proto = viaProto ? proto : builder.build();
|
||||
viaProto = true;
|
||||
return proto;
|
||||
}
|
||||
|
||||
private void mergeLocalToProto() {
|
||||
if (viaProto) {
|
||||
maybeInitBuilder();
|
||||
}
|
||||
mergeLocalToBuilder();
|
||||
proto = builder.build();
|
||||
viaProto = true;
|
||||
}
|
||||
|
||||
private void maybeInitBuilder() {
|
||||
if (viaProto || builder == null) {
|
||||
builder = RouterMasterKeyResponseProto.newBuilder(proto);
|
||||
}
|
||||
viaProto = false;
|
||||
}
|
||||
|
||||
private void mergeLocalToBuilder() {
|
||||
RouterMasterKeyPBImpl masterKeyRequest = (RouterMasterKeyPBImpl) this.routerMasterKey;
|
||||
RouterMasterKeyProto routerMasterKeyProto = builder.getRouterMasterKey();
|
||||
if (this.routerMasterKey != null && !masterKeyRequest.getProto().equals(routerMasterKeyProto)) {
|
||||
builder.setRouterMasterKey(convertToProtoFormat(this.routerMasterKey));
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public RouterMasterKey getRouterMasterKey() {
|
||||
RouterMasterKeyResponseProtoOrBuilder p = viaProto ? proto : builder;
|
||||
if (this.routerMasterKey != null) {
|
||||
return this.routerMasterKey;
|
||||
}
|
||||
if (!p.hasRouterMasterKey()) {
|
||||
return null;
|
||||
}
|
||||
this.routerMasterKey = convertFromProtoFormat(p.getRouterMasterKey());
|
||||
return this.routerMasterKey;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void setRouterMasterKey(RouterMasterKey masterKey) {
|
||||
maybeInitBuilder();
|
||||
if (masterKey == null) {
|
||||
builder.clearRouterMasterKey();
|
||||
}
|
||||
this.routerMasterKey = masterKey;
|
||||
}
|
||||
|
||||
@Override
|
||||
public int hashCode() {
|
||||
return getProto().hashCode();
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean equals(Object other) {
|
||||
if (other == null) {
|
||||
return false;
|
||||
}
|
||||
if (other.getClass().isAssignableFrom(this.getClass())) {
|
||||
return this.getProto().equals(this.getClass().cast(other).getProto());
|
||||
}
|
||||
return false;
|
||||
}
|
||||
|
||||
@Override
|
||||
public String toString() {
|
||||
return TextFormat.shortDebugString(getProto());
|
||||
}
|
||||
|
||||
private RouterMasterKey convertFromProtoFormat(RouterMasterKeyProto masterKeyProto) {
|
||||
return new RouterMasterKeyPBImpl(masterKeyProto);
|
||||
}
|
||||
|
||||
private RouterMasterKeyProto convertToProtoFormat(RouterMasterKey masterKey) {
|
||||
return ((RouterMasterKeyPBImpl) masterKey).getProto();
|
||||
}
|
||||
}
|
@ -18,6 +18,8 @@
|
||||
|
||||
package org.apache.hadoop.yarn.server.federation.utils;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.nio.ByteBuffer;
|
||||
import java.util.HashMap;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
@ -41,6 +43,7 @@
|
||||
import org.apache.hadoop.io.retry.RetryPolicies;
|
||||
import org.apache.hadoop.io.retry.RetryPolicy;
|
||||
import org.apache.hadoop.io.retry.RetryProxy;
|
||||
import org.apache.hadoop.security.token.delegation.DelegationKey;
|
||||
import org.apache.hadoop.util.ReflectionUtils;
|
||||
import org.apache.hadoop.yarn.api.records.ApplicationId;
|
||||
import org.apache.hadoop.yarn.api.records.ReservationId;
|
||||
@ -74,6 +77,9 @@
|
||||
import org.apache.hadoop.yarn.server.federation.store.records.UpdateApplicationHomeSubClusterRequest;
|
||||
import org.apache.hadoop.yarn.server.federation.store.records.UpdateReservationHomeSubClusterRequest;
|
||||
import org.apache.hadoop.yarn.server.federation.store.records.DeleteReservationHomeSubClusterRequest;
|
||||
import org.apache.hadoop.yarn.server.federation.store.records.RouterMasterKeyRequest;
|
||||
import org.apache.hadoop.yarn.server.federation.store.records.RouterMasterKeyResponse;
|
||||
import org.apache.hadoop.yarn.server.federation.store.records.RouterMasterKey;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
|
||||
@ -720,4 +726,56 @@ protected Object getAppHomeSubClusterCacheRequest(ApplicationId applicationId) {
|
||||
public FederationStateStore getStateStore() {
|
||||
return stateStore;
|
||||
}
|
||||
|
||||
/**
|
||||
* The Router Supports Store NewMasterKey (RouterMasterKey{@link RouterMasterKey}).
|
||||
*
|
||||
* @param newKey Key used for generating and verifying delegation tokens
|
||||
* @throws YarnException if the call to the state store is unsuccessful
|
||||
* @throws IOException An IO Error occurred
|
||||
* @return RouterMasterKeyResponse
|
||||
*/
|
||||
public RouterMasterKeyResponse storeNewMasterKey(DelegationKey newKey)
|
||||
throws YarnException, IOException {
|
||||
LOG.info("Storing master key with keyID {}.", newKey.getKeyId());
|
||||
ByteBuffer keyBytes = ByteBuffer.wrap(newKey.getEncodedKey());
|
||||
RouterMasterKey masterKey = RouterMasterKey.newInstance(newKey.getKeyId(),
|
||||
keyBytes, newKey.getExpiryDate());
|
||||
RouterMasterKeyRequest keyRequest = RouterMasterKeyRequest.newInstance(masterKey);
|
||||
return stateStore.storeNewMasterKey(keyRequest);
|
||||
}
|
||||
|
||||
/**
|
||||
* The Router Supports Remove MasterKey (RouterMasterKey{@link RouterMasterKey}).
|
||||
*
|
||||
* @param newKey Key used for generating and verifying delegation tokens
|
||||
* @throws YarnException if the call to the state store is unsuccessful
|
||||
* @throws IOException An IO Error occurred
|
||||
*/
|
||||
public void removeStoredMasterKey(DelegationKey newKey) throws YarnException, IOException {
|
||||
LOG.info("Removing master key with keyID {}.", newKey.getKeyId());
|
||||
ByteBuffer keyBytes = ByteBuffer.wrap(newKey.getEncodedKey());
|
||||
RouterMasterKey masterKey = RouterMasterKey.newInstance(newKey.getKeyId(),
|
||||
keyBytes, newKey.getExpiryDate());
|
||||
RouterMasterKeyRequest keyRequest = RouterMasterKeyRequest.newInstance(masterKey);
|
||||
stateStore.removeStoredMasterKey(keyRequest);
|
||||
}
|
||||
|
||||
/**
|
||||
* The Router Supports GetMasterKeyByDelegationKey.
|
||||
*
|
||||
* @param newKey Key used for generating and verifying delegation tokens
|
||||
* @throws YarnException if the call to the state store is unsuccessful
|
||||
* @throws IOException An IO Error occurred
|
||||
* @return RouterMasterKeyResponse
|
||||
*/
|
||||
public RouterMasterKeyResponse getMasterKeyByDelegationKey(DelegationKey newKey)
|
||||
throws YarnException, IOException {
|
||||
LOG.info("Storing master key with keyID {}.", newKey.getKeyId());
|
||||
ByteBuffer keyBytes = ByteBuffer.wrap(newKey.getEncodedKey());
|
||||
RouterMasterKey masterKey = RouterMasterKey.newInstance(newKey.getKeyId(),
|
||||
keyBytes, newKey.getExpiryDate());
|
||||
RouterMasterKeyRequest keyRequest = RouterMasterKeyRequest.newInstance(masterKey);
|
||||
return stateStore.getMasterKeyByDelegationKey(keyRequest);
|
||||
}
|
||||
}
|
||||
|
@ -25,6 +25,7 @@ package hadoop.yarn;
|
||||
|
||||
import "yarn_protos.proto";
|
||||
import "yarn_server_common_protos.proto";
|
||||
import "yarn_security_token.proto";
|
||||
|
||||
message SubClusterIdProto {
|
||||
optional string id = 1;
|
||||
@ -213,4 +214,31 @@ message GetSubClusterPoliciesConfigurationsRequestProto {
|
||||
|
||||
message GetSubClusterPoliciesConfigurationsResponseProto {
|
||||
repeated SubClusterPolicyConfigurationProto policies_configurations = 1;
|
||||
}
|
||||
}
|
||||
|
||||
message RouterMasterKeyProto {
|
||||
optional int32 key_id = 1;
|
||||
optional bytes key_bytes = 2;
|
||||
optional uint64 expiry_date = 3;
|
||||
}
|
||||
|
||||
message RouterMasterKeyRequestProto {
|
||||
optional RouterMasterKeyProto router_master_key = 1;
|
||||
}
|
||||
|
||||
message RouterMasterKeyResponseProto {
|
||||
optional RouterMasterKeyProto router_master_key = 1;
|
||||
}
|
||||
|
||||
message RouterStoreTokenProto {
|
||||
optional YARNDelegationTokenIdentifierProto token_identifier = 1;
|
||||
optional int64 renew_date = 2;
|
||||
}
|
||||
|
||||
message RouterRMTokenRequestProto {
|
||||
optional RouterStoreTokenProto router_store_token = 1;
|
||||
}
|
||||
|
||||
message RouterRMTokenResponseProto {
|
||||
optional RouterStoreTokenProto router_store_token = 1;
|
||||
}
|
@ -21,9 +21,12 @@
|
||||
import java.nio.ByteBuffer;
|
||||
import java.util.Calendar;
|
||||
import java.util.List;
|
||||
import java.util.Set;
|
||||
import java.util.HashSet;
|
||||
import java.util.TimeZone;
|
||||
|
||||
import org.apache.hadoop.conf.Configuration;
|
||||
import org.apache.hadoop.security.token.delegation.DelegationKey;
|
||||
import org.apache.hadoop.test.LambdaTestUtils;
|
||||
import org.apache.hadoop.util.Time;
|
||||
import org.apache.hadoop.yarn.api.records.ApplicationId;
|
||||
@ -68,6 +71,9 @@
|
||||
import org.apache.hadoop.yarn.server.federation.store.records.DeleteReservationHomeSubClusterResponse;
|
||||
import org.apache.hadoop.yarn.server.federation.store.records.UpdateReservationHomeSubClusterRequest;
|
||||
import org.apache.hadoop.yarn.server.federation.store.records.UpdateReservationHomeSubClusterResponse;
|
||||
import org.apache.hadoop.yarn.server.federation.store.records.RouterMasterKey;
|
||||
import org.apache.hadoop.yarn.server.federation.store.records.RouterMasterKeyRequest;
|
||||
import org.apache.hadoop.yarn.server.federation.store.records.RouterMasterKeyResponse;
|
||||
import org.apache.hadoop.yarn.util.MonotonicClock;
|
||||
import org.junit.After;
|
||||
import org.junit.Assert;
|
||||
@ -760,4 +766,75 @@ public void testUpdateReservationHomeSubClusterUnknownApp() throws Exception {
|
||||
"Reservation " + reservationId + " does not exist",
|
||||
() -> stateStore.updateReservationHomeSubCluster(updateReservationRequest));
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testStoreNewMasterKey() throws Exception {
|
||||
// store delegation key;
|
||||
DelegationKey key = new DelegationKey(1234, 4321, "keyBytes".getBytes());
|
||||
Set<DelegationKey> keySet = new HashSet<>();
|
||||
keySet.add(key);
|
||||
|
||||
RouterMasterKey routerMasterKey = RouterMasterKey.newInstance(key.getKeyId(),
|
||||
ByteBuffer.wrap(key.getEncodedKey()), key.getExpiryDate());
|
||||
RouterMasterKeyRequest routerMasterKeyRequest =
|
||||
RouterMasterKeyRequest.newInstance(routerMasterKey);
|
||||
RouterMasterKeyResponse response = stateStore.storeNewMasterKey(routerMasterKeyRequest);
|
||||
|
||||
Assert.assertNotNull(response);
|
||||
RouterMasterKey routerMasterKeyResp = response.getRouterMasterKey();
|
||||
Assert.assertNotNull(routerMasterKeyResp);
|
||||
Assert.assertEquals(routerMasterKey.getKeyId(), routerMasterKeyResp.getKeyId());
|
||||
Assert.assertEquals(routerMasterKey.getKeyBytes(), routerMasterKeyResp.getKeyBytes());
|
||||
Assert.assertEquals(routerMasterKey.getExpiryDate(), routerMasterKeyResp.getExpiryDate());
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testGetMasterKeyByDelegationKey() throws YarnException, IOException {
|
||||
// store delegation key;
|
||||
DelegationKey key = new DelegationKey(5678, 8765, "keyBytes".getBytes());
|
||||
Set<DelegationKey> keySet = new HashSet<>();
|
||||
keySet.add(key);
|
||||
|
||||
RouterMasterKey routerMasterKey = RouterMasterKey.newInstance(key.getKeyId(),
|
||||
ByteBuffer.wrap(key.getEncodedKey()), key.getExpiryDate());
|
||||
RouterMasterKeyRequest routerMasterKeyRequest =
|
||||
RouterMasterKeyRequest.newInstance(routerMasterKey);
|
||||
RouterMasterKeyResponse response = stateStore.storeNewMasterKey(routerMasterKeyRequest);
|
||||
Assert.assertNotNull(response);
|
||||
|
||||
RouterMasterKeyResponse routerMasterKeyResponse =
|
||||
stateStore.getMasterKeyByDelegationKey(routerMasterKeyRequest);
|
||||
|
||||
Assert.assertNotNull(routerMasterKeyResponse);
|
||||
|
||||
RouterMasterKey routerMasterKeyResp = routerMasterKeyResponse.getRouterMasterKey();
|
||||
Assert.assertNotNull(routerMasterKeyResp);
|
||||
Assert.assertEquals(routerMasterKey.getKeyId(), routerMasterKeyResp.getKeyId());
|
||||
Assert.assertEquals(routerMasterKey.getKeyBytes(), routerMasterKeyResp.getKeyBytes());
|
||||
Assert.assertEquals(routerMasterKey.getExpiryDate(), routerMasterKeyResp.getExpiryDate());
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testRemoveStoredMasterKey() throws YarnException, IOException {
|
||||
// store delegation key;
|
||||
DelegationKey key = new DelegationKey(1234, 4321, "keyBytes".getBytes());
|
||||
Set<DelegationKey> keySet = new HashSet<>();
|
||||
keySet.add(key);
|
||||
|
||||
RouterMasterKey routerMasterKey = RouterMasterKey.newInstance(key.getKeyId(),
|
||||
ByteBuffer.wrap(key.getEncodedKey()), key.getExpiryDate());
|
||||
RouterMasterKeyRequest routerMasterKeyRequest =
|
||||
RouterMasterKeyRequest.newInstance(routerMasterKey);
|
||||
RouterMasterKeyResponse response = stateStore.storeNewMasterKey(routerMasterKeyRequest);
|
||||
Assert.assertNotNull(response);
|
||||
|
||||
RouterMasterKeyResponse masterKeyResponse =
|
||||
stateStore.removeStoredMasterKey(routerMasterKeyRequest);
|
||||
Assert.assertNotNull(masterKeyResponse);
|
||||
|
||||
RouterMasterKey routerMasterKeyResp = masterKeyResponse.getRouterMasterKey();
|
||||
Assert.assertEquals(routerMasterKey.getKeyId(), routerMasterKeyResp.getKeyId());
|
||||
Assert.assertEquals(routerMasterKey.getKeyBytes(), routerMasterKeyResp.getKeyBytes());
|
||||
Assert.assertEquals(routerMasterKey.getExpiryDate(), routerMasterKeyResp.getExpiryDate());
|
||||
}
|
||||
}
|
||||
|
@ -17,6 +17,7 @@
|
||||
|
||||
package org.apache.hadoop.yarn.server.federation.store.impl;
|
||||
|
||||
import org.apache.commons.lang3.NotImplementedException;
|
||||
import org.apache.hadoop.test.LambdaTestUtils;
|
||||
import org.apache.hadoop.util.Time;
|
||||
import org.apache.hadoop.yarn.api.records.ApplicationId;
|
||||
@ -555,4 +556,19 @@ public void testDeleteReservationHomeSubClusterAbnormalSituation() throws Except
|
||||
LambdaTestUtils.intercept(YarnException.class, errorMsg,
|
||||
() -> stateStore.deleteReservationHomeSubCluster(delRequest));
|
||||
}
|
||||
|
||||
@Test(expected = NotImplementedException.class)
|
||||
public void testStoreNewMasterKey() throws Exception {
|
||||
super.testStoreNewMasterKey();
|
||||
}
|
||||
|
||||
@Test(expected = NotImplementedException.class)
|
||||
public void testGetMasterKeyByDelegationKey() throws YarnException, IOException {
|
||||
super.testGetMasterKeyByDelegationKey();
|
||||
}
|
||||
|
||||
@Test(expected = NotImplementedException.class)
|
||||
public void testRemoveStoredMasterKey() throws YarnException, IOException {
|
||||
super.testRemoveStoredMasterKey();
|
||||
}
|
||||
}
|
@ -19,6 +19,7 @@
|
||||
|
||||
import java.io.IOException;
|
||||
|
||||
import org.apache.commons.lang3.NotImplementedException;
|
||||
import org.apache.curator.framework.CuratorFramework;
|
||||
import org.apache.curator.framework.CuratorFrameworkFactory;
|
||||
import org.apache.curator.retry.RetryNTimes;
|
||||
@ -168,4 +169,19 @@ public void testMetricsInited() throws Exception {
|
||||
MetricsRecords.assertMetric(record, "DeleteReservationHomeSubClusterNumOps", expectOps);
|
||||
MetricsRecords.assertMetric(record, "UpdateReservationHomeSubClusterNumOps", expectOps);
|
||||
}
|
||||
|
||||
@Test(expected = NotImplementedException.class)
|
||||
public void testStoreNewMasterKey() throws Exception {
|
||||
super.testStoreNewMasterKey();
|
||||
}
|
||||
|
||||
@Test(expected = NotImplementedException.class)
|
||||
public void testGetMasterKeyByDelegationKey() throws YarnException, IOException {
|
||||
super.testGetMasterKeyByDelegationKey();
|
||||
}
|
||||
|
||||
@Test(expected = NotImplementedException.class)
|
||||
public void testRemoveStoredMasterKey() throws YarnException, IOException {
|
||||
super.testRemoveStoredMasterKey();
|
||||
}
|
||||
}
|
@ -47,6 +47,9 @@
|
||||
import org.apache.hadoop.yarn.federation.proto.YarnServerFederationProtos.SubClusterRegisterResponseProto;
|
||||
import org.apache.hadoop.yarn.federation.proto.YarnServerFederationProtos.UpdateApplicationHomeSubClusterRequestProto;
|
||||
import org.apache.hadoop.yarn.federation.proto.YarnServerFederationProtos.UpdateApplicationHomeSubClusterResponseProto;
|
||||
import org.apache.hadoop.yarn.federation.proto.YarnServerFederationProtos.RouterMasterKeyProto;
|
||||
import org.apache.hadoop.yarn.federation.proto.YarnServerFederationProtos.RouterMasterKeyRequestProto;
|
||||
import org.apache.hadoop.yarn.federation.proto.YarnServerFederationProtos.RouterMasterKeyResponseProto;
|
||||
import org.apache.hadoop.yarn.server.federation.store.records.impl.pb.AddApplicationHomeSubClusterRequestPBImpl;
|
||||
import org.apache.hadoop.yarn.server.federation.store.records.impl.pb.AddApplicationHomeSubClusterResponsePBImpl;
|
||||
import org.apache.hadoop.yarn.server.federation.store.records.impl.pb.DeleteApplicationHomeSubClusterRequestPBImpl;
|
||||
@ -75,6 +78,9 @@
|
||||
import org.apache.hadoop.yarn.server.federation.store.records.impl.pb.SubClusterRegisterResponsePBImpl;
|
||||
import org.apache.hadoop.yarn.server.federation.store.records.impl.pb.UpdateApplicationHomeSubClusterRequestPBImpl;
|
||||
import org.apache.hadoop.yarn.server.federation.store.records.impl.pb.UpdateApplicationHomeSubClusterResponsePBImpl;
|
||||
import org.apache.hadoop.yarn.server.federation.store.records.impl.pb.RouterMasterKeyPBImpl;
|
||||
import org.apache.hadoop.yarn.server.federation.store.records.impl.pb.RouterMasterKeyRequestPBImpl;
|
||||
import org.apache.hadoop.yarn.server.federation.store.records.impl.pb.RouterMasterKeyResponsePBImpl;
|
||||
import org.apache.hadoop.yarn.server.records.Version;
|
||||
import org.junit.BeforeClass;
|
||||
import org.junit.Test;
|
||||
@ -92,6 +98,7 @@ public static void setup() throws Exception {
|
||||
generateByNewInstance(SubClusterInfo.class);
|
||||
generateByNewInstance(ApplicationHomeSubCluster.class);
|
||||
generateByNewInstance(SubClusterPolicyConfiguration.class);
|
||||
generateByNewInstance(RouterMasterKey.class);
|
||||
}
|
||||
|
||||
@Test
|
||||
@ -262,4 +269,19 @@ public void testGetSubClusterPoliciesConfigurationsResponse()
|
||||
GetSubClusterPoliciesConfigurationsResponsePBImpl.class,
|
||||
GetSubClusterPoliciesConfigurationsResponseProto.class);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testRouterMasterKey() throws Exception {
|
||||
validatePBImplRecord(RouterMasterKeyPBImpl.class, RouterMasterKeyProto.class);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testRouterMasterKeyRequest() throws Exception {
|
||||
validatePBImplRecord(RouterMasterKeyRequestPBImpl.class, RouterMasterKeyRequestProto.class);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testRouterMasterKeyResponse() throws Exception {
|
||||
validatePBImplRecord(RouterMasterKeyResponsePBImpl.class, RouterMasterKeyResponseProto.class);
|
||||
}
|
||||
}
|
||||
|
@ -22,8 +22,11 @@
|
||||
import java.util.Arrays;
|
||||
import java.util.Collection;
|
||||
import java.util.Map;
|
||||
import java.util.Set;
|
||||
import java.util.HashSet;
|
||||
|
||||
import org.apache.hadoop.conf.Configuration;
|
||||
import org.apache.hadoop.security.token.delegation.DelegationKey;
|
||||
import org.apache.hadoop.yarn.api.records.ApplicationId;
|
||||
import org.apache.hadoop.yarn.conf.YarnConfiguration;
|
||||
import org.apache.hadoop.yarn.exceptions.YarnException;
|
||||
@ -33,6 +36,7 @@
|
||||
import org.apache.hadoop.yarn.server.federation.store.records.SubClusterId;
|
||||
import org.apache.hadoop.yarn.server.federation.store.records.SubClusterInfo;
|
||||
import org.apache.hadoop.yarn.server.federation.store.records.SubClusterPolicyConfiguration;
|
||||
import org.apache.hadoop.yarn.server.federation.store.records.RouterRMDTSecretManagerState;
|
||||
import org.junit.After;
|
||||
import org.junit.Assert;
|
||||
import org.junit.Before;
|
||||
@ -232,4 +236,37 @@ public void testGetApplicationHomeSubClusterCache() throws YarnException {
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testStoreNewMasterKey() throws YarnException, IOException {
|
||||
// store delegation key;
|
||||
DelegationKey key = new DelegationKey(1234, 4321, "keyBytes".getBytes());
|
||||
Set<DelegationKey> keySet = new HashSet<>();
|
||||
keySet.add(key);
|
||||
facade.storeNewMasterKey(key);
|
||||
|
||||
MemoryFederationStateStore federationStateStore =
|
||||
(MemoryFederationStateStore) facade.getStateStore();
|
||||
RouterRMDTSecretManagerState secretManagerState =
|
||||
federationStateStore.getRouterRMSecretManagerState();
|
||||
Assert.assertEquals(keySet, secretManagerState.getMasterKeyState());
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testRemoveStoredMasterKey() throws YarnException, IOException {
|
||||
// store delegation key;
|
||||
DelegationKey key = new DelegationKey(4567, 7654, "keyBytes".getBytes());
|
||||
Set<DelegationKey> keySet = new HashSet<>();
|
||||
keySet.add(key);
|
||||
facade.storeNewMasterKey(key);
|
||||
|
||||
// check to delete delegationKey
|
||||
facade.removeStoredMasterKey(key);
|
||||
keySet.clear();
|
||||
|
||||
MemoryFederationStateStore federationStateStore =
|
||||
(MemoryFederationStateStore) facade.getStateStore();
|
||||
RouterRMDTSecretManagerState secretManagerState =
|
||||
federationStateStore.getRouterRMSecretManagerState();
|
||||
Assert.assertEquals(keySet, secretManagerState.getMasterKeyState());
|
||||
}
|
||||
}
|
||||
|
@ -18,10 +18,12 @@
|
||||
|
||||
package org.apache.hadoop.yarn.server.resourcemanager.federation;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.net.InetSocketAddress;
|
||||
import java.util.concurrent.ScheduledExecutorService;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
|
||||
import org.apache.commons.lang3.NotImplementedException;
|
||||
import org.apache.hadoop.conf.Configuration;
|
||||
import org.apache.hadoop.io.retry.RetryPolicy;
|
||||
import org.apache.hadoop.net.NetUtils;
|
||||
@ -70,6 +72,8 @@
|
||||
import org.apache.hadoop.yarn.server.federation.store.records.UpdateApplicationHomeSubClusterResponse;
|
||||
import org.apache.hadoop.yarn.server.federation.store.records.UpdateReservationHomeSubClusterRequest;
|
||||
import org.apache.hadoop.yarn.server.federation.store.records.UpdateReservationHomeSubClusterResponse;
|
||||
import org.apache.hadoop.yarn.server.federation.store.records.RouterMasterKeyRequest;
|
||||
import org.apache.hadoop.yarn.server.federation.store.records.RouterMasterKeyResponse;
|
||||
import org.apache.hadoop.yarn.server.federation.utils.FederationStateStoreFacade;
|
||||
import org.apache.hadoop.yarn.server.records.Version;
|
||||
import org.apache.hadoop.yarn.server.resourcemanager.RMContext;
|
||||
@ -356,4 +360,22 @@ public DeleteReservationHomeSubClusterResponse deleteReservationHomeSubCluster(
|
||||
DeleteReservationHomeSubClusterRequest request) throws YarnException {
|
||||
return stateStoreClient.deleteReservationHomeSubCluster(request);
|
||||
}
|
||||
|
||||
@Override
|
||||
public RouterMasterKeyResponse storeNewMasterKey(RouterMasterKeyRequest request)
|
||||
throws YarnException, IOException {
|
||||
throw new NotImplementedException("Code is not implemented");
|
||||
}
|
||||
|
||||
@Override
|
||||
public RouterMasterKeyResponse removeStoredMasterKey(RouterMasterKeyRequest request)
|
||||
throws YarnException, IOException {
|
||||
throw new NotImplementedException("Code is not implemented");
|
||||
}
|
||||
|
||||
@Override
|
||||
public RouterMasterKeyResponse getMasterKeyByDelegationKey(RouterMasterKeyRequest request)
|
||||
throws YarnException, IOException {
|
||||
throw new NotImplementedException("Code is not implemented");
|
||||
}
|
||||
}
|
||||
|
Loading…
Reference in New Issue
Block a user