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

[stress] Sync namespace federated credentials periodically and on startup #9063

Merged
merged 4 commits into from
Oct 3, 2024
Merged

[stress] Sync namespace federated credentials periodically and on startup #9063

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
411d4ff
Watcher namespace fed cred management fixes
benbp Sep 26, 2024
c00d5c7
Update stress watcher dockerfile to use net 8
benbp Sep 26, 2024
ed1b098
Add back delete code, remove terminating skip
benbp Sep 27, 2024
89ea415
Bump cluster version to 1.29.8
benbp Oct 3, 2024
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
133 changes: 78 additions & 55 deletions tools/stress-cluster/services/Stress.Watcher/src/NamespaceEventHandler.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,6 +28,8 @@ public class NamespaceEventHandler
// Concurrent Federated Identity Credentials writes under the same managed identity are not supported
private static readonly SemaphoreSlim FederatedCredentialWriteSemaphore = new(1, 1);

private Dictionary<string, UserAssignedIdentityResource> WorkloadAppCache = [];

public List<string> WorkloadAppPool;
public string WorkloadAppIssuer;

Expand Down Expand Up @@ -62,6 +64,57 @@ public NamespaceEventHandler(
.CreateLogger();
}

public async Task SyncCredentials()
{
try
{
Logger.Information($"Waiting for federated credential write semaphore");
await FederatedCredentialWriteSemaphore.WaitAsync();
await _syncCredentials();
}
finally
{
Logger.Information("Releasing federated credential write semaphore");
FederatedCredentialWriteSemaphore.Release();
}
}

public async Task _syncCredentials()
{
Logger.Information("Syncing namespaced federated credentials, this may take a minute...");

var namespaces = await Client.ListNamespaceAsync();
foreach (var app in WorkloadAppPool)
{
var resourceId = UserAssignedIdentityResource.CreateResourceIdentifier(SubscriptionId, ClusterGroup, app);
var userAssignedIdentity = ArmClient.GetUserAssignedIdentityResource(resourceId);
var identityResource = await userAssignedIdentity.GetAsync();
var fedCreds = userAssignedIdentity.GetFederatedIdentityCredentials();
await foreach (var item in fedCreds.GetAllAsync())
{
if (!namespaces.Items.Any(ns => item.Data.Name == CreateFederatedIdentityCredentialName(ns)))
{
if (!string.IsNullOrEmpty(WatchNamespace) && item.Data.Name != CreateFederatedIdentityCredentialName(WatchNamespace))
{
Logger.Information($"Skipping delete federated credential '{item.Data.Name}' because it is not the watched namespace '{WatchNamespace}'");
continue;
}
// Only perform delete operations for namespace state that may have changed if the watcher was not running.
// Any create operations will be handled after initialization as the watch stream processes all active namespaces on startup
Logger.Information($"Deleting federated identity credential '{item.Data.Name}' for managed identity '{app}' as the corresponding namespace no longer exists.");
WorkloadAppCache.Remove(item.Data.Name);
var lro = await item.DeleteAsync(Azure.WaitUntil.Completed);
}
else
{
WorkloadAppCache[item.Data.Name] = identityResource.Value;
}
}
}

Logger.Information($"Federated credential sync complete. Cached {WorkloadAppCache.Count} federated credentials.");
}

public async Task Watch(CancellationToken cancellationToken)
{
string resourceVersion = null;
Expand Down Expand Up @@ -125,6 +178,10 @@ public void HandleNamespaceEvent(WatchEventType eventType, V1Namespace ns)
Logger.Information($"Skipping namespace '{ns.Name()}' because it is not the watched namespace '{WatchNamespace}'");
return;
}
if (ns.Status?.Phase == "Terminating")
{
return;
}

using (LogContext.PushProperty("namespace", ns.Name()))
{
Expand All @@ -139,24 +196,17 @@ public void HandleNamespaceEvent(WatchEventType eventType, V1Namespace ns)
}
});
}
else if (eventType == WatchEventType.Deleted)
{
DeleteFederatedIdentityCredential(ns).ContinueWith(t =>
{
Logger.Information("Releasing federated credential write semaphore");
FederatedCredentialWriteSemaphore.Release();
if (t.Exception != null)
{
Logger.Error(t.Exception, "Error deleting federated identity credential.");
}
});
}
}
}

public string CreateFederatedIdentityCredentialName(V1Namespace ns)
{
return $"stress-{ns.Name()}";
return CreateFederatedIdentityCredentialName(ns.Name());
}

public string CreateFederatedIdentityCredentialName(string ns)
{
return $"stress-{ns}";
}

public async Task InitializeWorkloadIdForNamespace(V1Namespace ns)
Expand All @@ -175,14 +225,21 @@ public async Task InitializeWorkloadIdForNamespace(V1Namespace ns)
var identityData = await selectedWorkloadIdentity.GetAsync();
var selectedWorkloadAppId = identityData.Value.Data.ClientId.ToString();

var meta = new V1ObjectMeta(){
var meta = new V1ObjectMeta()
{
Name = ns.Name(),
NamespaceProperty = ns.Name(),
Annotations = new Dictionary<string, string>(){
{ "azure.workload.identity/client-id", selectedWorkloadAppId }
}
};
var serviceAccount = new V1ServiceAccount(metadata: meta);
var allAccounts = await Client.ListNamespacedServiceAccountAsync(ns.Name());
if (allAccounts.Items.Any(sa => sa.Name() == ns.Name()))
{
Logger.Information($"Service account '{ns.Name()}/{ns.Name()}' already exists, skipping creation.");
return;
}
await Client.CreateNamespacedServiceAccountAsync(serviceAccount, ns.Name());
Logger.Information($"Created service account '{ns.Name()}/{ns.Name()}' with workload client id '{selectedWorkloadAppId}'");
}
Expand All @@ -200,6 +257,12 @@ public async Task<UserAssignedIdentityResource> CreateFederatedIdentityCredentia
Logger.Information($"Waiting for federated credential write semaphore");
await FederatedCredentialWriteSemaphore.WaitAsync();

if (WorkloadAppCache.ContainsKey(credentialName))
{
Logger.Information($"Found cache entry for federated credential {credentialName}, returning identity {WorkloadAppCache[credentialName].Data.ClientId}");
return await WorkloadAppCache[credentialName].GetAsync();
}

foreach (var workloadApp in WorkloadAppPool)
{
var userAssignedIdentityResourceId = UserAssignedIdentityResource.CreateResourceIdentifier(SubscriptionId, ClusterGroup, workloadApp);
Expand Down Expand Up @@ -246,50 +309,10 @@ public async Task<UserAssignedIdentityResource> CreateFederatedIdentityCredentia
Logger.Information($"Creating/updating federated identity credential '{credentialName}' " +
$"with subject '{subject}' for managed identity '{selectedWorkloadApp}'");
var lro = await federatedIdentityCredential.UpdateAsync(Azure.WaitUntil.Completed, fedCredData);
WorkloadAppCache[credentialName] = selectedIdentity;
Logger.Information($"Created federated identity credential '{lro.Value.Data.Name}'");

return selectedIdentity;
}

public async Task DeleteFederatedIdentityCredential(V1Namespace ns)
{
var credentialName = CreateFederatedIdentityCredentialName(ns);
var workloadApp = "";
foreach (var app in WorkloadAppPool)
{
var resourceId = UserAssignedIdentityResource.CreateResourceIdentifier(SubscriptionId, ClusterGroup, app);
var userAssignedIdentity = ArmClient.GetUserAssignedIdentityResource(resourceId);
var fedCreds = userAssignedIdentity.GetFederatedIdentityCredentials();
await foreach (var item in fedCreds.GetAllAsync())
{
if (item.Data.Name == credentialName)
{
workloadApp = app;
break;
}
}
if (!String.IsNullOrEmpty(workloadApp))
{
break;
}
}

if (string.IsNullOrEmpty(workloadApp))
{
Logger.Warning($"Federated identity credential '{credentialName}' not found in workload app pool. Skipping delete.");
return;
}

var federatedIdentityCredentialResourceId = FederatedIdentityCredentialResource.CreateResourceIdentifier(
SubscriptionId, ClusterGroup, workloadApp, credentialName);
var federatedIdentityCredential = ArmClient.GetFederatedIdentityCredentialResource(federatedIdentityCredentialResourceId);

Logger.Information($"Waiting for federated credential write semaphore");
await FederatedCredentialWriteSemaphore.WaitAsync();

Logger.Information($"Deleting federated identity credential '{credentialName}' for managed identity '{workloadApp}'");
var lro = await federatedIdentityCredential.DeleteAsync(Azure.WaitUntil.Completed);
Logger.Information($"Deleted federated identity credential '{credentialName}'");
}
}
}
11 changes: 11 additions & 0 deletions tools/stress-cluster/services/Stress.Watcher/src/Program.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -79,6 +79,8 @@ static async Task Program(Options options)
var namespaceEventHandler = new NamespaceEventHandler(
client, armClient, workloadConfig.SubscriptionId, workloadConfig.ClusterGroup,
workloadConfig.WorkloadAppPool, workloadConfig.WorkloadAppIssuer, options.Namespace);
await namespaceEventHandler.SyncCredentials();
_ = PollAndSyncCredentials(namespaceEventHandler, 288); // poll every 12 hours

var cts = new CancellationTokenSource();
var taskList = new List<Task>
Expand Down Expand Up @@ -164,5 +166,14 @@ static WorkloadAuthConfig GetWorkloadConfigValues(Options options, Boolean isLoc
ClusterGroup = clusterGroup
};
}

static async Task PollAndSyncCredentials(NamespaceEventHandler namespaceHandler, int minutes)
{
while (true)
{
await Task.Delay(TimeSpan.FromMinutes(minutes));
await namespaceHandler.SyncCredentials();
}
}
}
}
Loading