Changed K8s provider to use annotations instead of labels for kind propagation

benchmarks/KubernetesDiagnostics/KubernetesDiagnostics.csproj

@@ -1,20 +1,18 @@
-<Project Sdk="Microsoft.NET.Sdk">
+<Project Sdk="Microsoft.NET.Sdk.Web">
 
     <PropertyGroup>
         <OutputType>Exe</OutputType>
-        <TargetFrameworks>net6.0;net7.0</TargetFrameworks>
+        <TargetFramework>net7.0</TargetFramework>
         <DockerDefaultTargetOS>Linux</DockerDefaultTargetOS>
     </PropertyGroup>
 
     <ItemGroup>
-      <PackageReference Include="KubernetesClient" Version="7.1.3-g341580fb9b" />
       <PackageReference Include="Microsoft.Extensions.Logging.Console" Version="6.0.0" />
-      <PackageReference Include="MongoDB.Driver" Version="2.17.1" />
-      <PackageReference Include="Proto.Cluster.Identity.MongoDb" Version="1.0.0-rc2.43" />
-        <PackageReference Include="Proto.Cluster.Identity.Redis" Version="1.0.0-rc2.43" />
-      <PackageReference Include="Proto.Cluster.Kubernetes" Version="1.0.0-rc2.43" />
-        <PackageReference Include="Proto.Cluster.Consul" Version="1.0.0-rc2.43" />
-      <PackageReference Include="StackExchange.Redis" Version="2.6.66" />
+      <PackageReference Include="Microsoft.NET.Build.Containers" Version="0.1.8" />
+    </ItemGroup>
+
+    <ItemGroup>
+      <ProjectReference Include="..\..\src\Proto.Cluster.Kubernetes\Proto.Cluster.Kubernetes.csproj" />
     </ItemGroup>
 
 </Project>

benchmarks/KubernetesDiagnostics/Program.cs

@@ -2,19 +2,14 @@
 using System.Linq;
 using System.Threading;
 using System.Threading.Tasks;
-using k8s;
 using Microsoft.Extensions.Logging;
 using Proto;
 using Proto.Cluster;
-using Proto.Cluster.Consul;
 using Proto.Cluster.Gossip;
-using Proto.Cluster.Identity;
-using Proto.Cluster.Identity.Redis;
 using Proto.Cluster.Kubernetes;
 using Proto.Cluster.Partition;
 using Proto.Remote;
 using Proto.Remote.GrpcNet;
-using StackExchange.Redis;
 
 namespace KubernetesDiagnostics;
 
@@ -24,7 +19,6 @@ public static async Task Main()
     {
         ThreadPool.SetMinThreads(100, 100);
         Console.WriteLine("Starting...");
-        Console.WriteLine("444");
 
         /*
          *  docker build . -t rogeralsing/kubdiagg   
@@ -38,9 +32,9 @@ public static async Task Main()
         Log.SetLoggerFactory(l);
         var log = Log.CreateLogger("main");
 
-        var identity = new PartitionIdentityLookup(TimeSpan.FromSeconds(2),TimeSpan.FromSeconds(2));//  new IdentityStorageLookup(GetRedisId("MyCluster"));
+        var identity = new PartitionIdentityLookup(TimeSpan.FromSeconds(2), TimeSpan.FromSeconds(2)
+        ); //  new IdentityStorageLookup(GetRedisId("MyCluster"));
 
-
         /*
         - name: "REDIS"
           value: "redis"
@@ -59,8 +53,10 @@ public static async Task Main()
         log.LogInformation("Port {port}", port);
         log.LogInformation("Advertised Host {advertisedHost}", advertisedHost);
 
-        var clusterprovider = GetProvider();
+        var clusterProvider = GetProvider();
 
+        var noOpsProps = Props.FromFunc(ctx => Task.CompletedTask);
+        var echoKind = new ClusterKind("echo", noOpsProps);
         var system = new ActorSystem(new ActorSystemConfig()
                 //     .WithDeveloperReceiveLogging(TimeSpan.FromSeconds(1))
                 //     .WithDeveloperSupervisionLogging(true)
@@ -71,15 +67,14 @@ public static async Task Main()
                 .WithEndpointWriterMaxRetries(2)
             )
             .WithCluster(ClusterConfig
-                .Setup("mycluster", clusterprovider, identity)
+                .Setup("mycluster", clusterProvider, identity)
                 .WithClusterKind("empty", Props.Empty)
+                .WithClusterKind(echoKind)
             );
 
-        system.EventStream.Subscribe<GossipUpdate>(e => {
-                Console.WriteLine($"{DateTime.Now:O} Gossip update Member {e.MemberId} Key {e.Key}");
-            }
+        system.EventStream.Subscribe<GossipUpdate>(e => { Console.WriteLine($"{DateTime.Now:O} Gossip update Member {e.MemberId} Key {e.Key}"); }
         );
-            
+
         system.EventStream.Subscribe<ClusterTopology>(e => {
                 var members = e.Members;
                 var x = members.Select(m => m.Id).OrderBy(i => i).ToArray();
@@ -95,26 +90,49 @@ public static async Task Main()
             }
         );
 
+        var cts = new CancellationTokenSource();
+
+        Console.CancelKeyPress += (_, _) => { cts.Cancel(); };
+
         await system
             .Cluster()
             .StartMemberAsync();
 
         var props = Props.FromFunc(ctx => Task.CompletedTask);
         system.Root.SpawnNamed(props, "dummy");
 
+        var clusterIdentity = ClusterIdentity.Create("some-id", echoKind.Name);
 
-        while (true)
+        while (!cts.IsCancellationRequested)
         {
-            var res = await system.Cluster().MemberList.TopologyConsensus(CancellationTokens.FromSeconds(5));
+            // var res = await system.Cluster().MemberList.TopologyConsensus(CancellationTokens.FromSeconds(5));
 
             var m = system.Cluster().MemberList.GetAllMembers();
             var hash = Member.TopologyHash(m);
-
-            Console.WriteLine($"{DateTime.Now:O} Consensus {res}.. Hash {hash} Count {m.Length}");
+
+            Console.WriteLine($"{DateTime.Now:O} Hash {hash} Count {m.Length}");
+
+            try
+            {
+                var t = await system.Cluster().RequestAsync<Touched>(clusterIdentity, new Touch(), CancellationTokens.FromSeconds(1));
+
+                if (t != null)
+                {
+                    Console.WriteLine($"called cluster actor {t.Who}");
+                }
+                else
+                {
+                    Console.WriteLine($"call to cluster actor returned null");
+                }
+            }
+            catch (Exception e)
+            {
+                Console.WriteLine($"Could not call cluster actor: {e}");
+            }
 
             foreach (var member in m)
             {
-                var pid = new PID(member.Address,"dummy");
+                var pid = new PID(member.Address, "dummy");
 
                 try
                 {
@@ -137,31 +155,11 @@ await system
 
             await Task.Delay(3000);
         }
+
+        await system
+            .Cluster()
+            .ShutdownAsync();
     }
 
-    private static IClusterProvider GetProvider()
-    {
-        try
-        {
-            Console.WriteLine("Running with Kubernetes Provider");
-            return new KubernetesProvider(new KubernetesProviderConfig());
-        }
-        catch
-        {
-            Console.WriteLine("Running with Consul Provider");
-            return new ConsulProvider(new ConsulProviderConfig());
-        }
-    }
-
-    private static IIdentityStorage GetRedisId(string clusterName)
-    {
-        var connectionString =
-            Environment.GetEnvironmentVariable("REDIS");
-
-        Console.WriteLine("REDIS " + connectionString);
-
-        var multiplexer = ConnectionMultiplexer.Connect(connectionString);
-        var identity = new RedisIdentityStorage(clusterName, multiplexer);
-        return identity;
-    }
+    private static IClusterProvider GetProvider() => new KubernetesProvider();
 }

benchmarks/KubernetesDiagnostics/build.sh

@@ -1,4 +1,3 @@
 kubectl delete --filename service.yaml
-docker build . -t rogeralsing/kubdiagg:default2 -f Dockerfile
-#docker push docker.io/rogeralsing/kubdiagg:default
+dotnet publish --os linux -c Release --arch x64 -p:PublishProfile=DefaultContainer
 kubectl apply --filename service.yaml

src/Proto.Cluster.Kubernetes/KubernetesExtensions.cs

@@ -31,17 +31,20 @@ internal static class KubernetesExtensions
     /// <param name="podNamespace">Namespace of the pod</param>
     /// <param name="pod">the pod that should be patched</param>
     /// <param name="labels">Labels collection. All labels will be replaced by the new labels.</param>
+    /// <param name="annotations">annotations collection. All annotations will be replaced by the new annotations.</param>
     /// <returns></returns>
-    internal static Task<V1Pod> ReplacePodLabels(
+    internal static Task<V1Pod> ReplacePodLabelsAndAnnotations(
         this IKubernetes kubernetes,
         string podName,
         string podNamespace,
         V1Pod pod,
-        IDictionary<string, string> labels
+        IDictionary<string, string> labels,
+        IDictionary<string, string> annotations
     )
     {
         var old = JsonSerializer.SerializeToDocument(pod);
         pod.Metadata.Labels = labels;
+        pod.Metadata.Annotations = annotations;
         var expected = JsonSerializer.SerializeToDocument(pod);
         var patch = old.CreatePatch(expected);
 
@@ -60,9 +63,9 @@ internal static (bool IsCandidate, bool IsAlive, Member Status) GetMemberStatus(
 
         var kinds = pod
             .Metadata
-            .Labels
-            .Where(l => l.Key.StartsWith(LabelKind) && l.Value == "true")
-            .Select(l => l.Key.Substring(LabelKind.Length + 1))
+            .Annotations
+            .Where(l => l.Key.StartsWith(AnnotationKinds))
+            .SelectMany(l => l.Value.Split(';'))
             .ToArray();
 
         var host = pod.Status.PodIP ?? "";

src/Proto.Cluster.Kubernetes/KubernetesProvider.cs

@@ -6,6 +6,7 @@
 
 using System;
 using System.Collections.Generic;
+using System.Linq;
 using System.Threading.Tasks;
 using JetBrains.Annotations;
 using k8s;
@@ -126,30 +127,34 @@ public async Task RegisterMemberInner()
 
         Logger.LogInformation("[Cluster][KubernetesProvider] Using Kubernetes port: {Port}", _port);
 
-        var existingLabels = pod.Metadata.Labels;
-
         var labels = new Dictionary<string, string>
         {
             [LabelCluster] = _clusterName,
             [LabelPort] = _port.ToString(),
             [LabelMemberId] = _cluster.System.Id
         };
 
-        foreach (var kind in _kinds)
+        foreach (var existing in pod.Metadata.Labels)
         {
-            var labelKey = $"{LabelKind}-{kind}";
-            labels.TryAdd(labelKey, "true");
+            labels.TryAdd(existing.Key, existing.Value);
         }
 
-        //add existing labels back
-        foreach (var existing in existingLabels)
+        var annotations = new Dictionary<string, string>
         {
-            labels.TryAdd(existing.Key, existing.Value);
+            [AnnotationKinds] = string.Join(';', _kinds),
+        };
+
+        if (pod.Metadata.Annotations is not null)
+        {
+            foreach (var existing in pod.Metadata.Annotations)
+            {
+                annotations.TryAdd(existing.Key, existing.Value);
+            }
         }
 
         try
         {
-            await kubernetes.ReplacePodLabels(_podName, KubernetesExtensions.GetKubeNamespace(), pod, labels);
+            await kubernetes.ReplacePodLabelsAndAnnotations(_podName, KubernetesExtensions.GetKubeNamespace(), pod, labels, annotations);
         }
         catch (Exception e)
         {
@@ -204,24 +209,15 @@ private async Task DeregisterMemberInner(Cluster cluster)
 
         var pod = await kubernetes.ReadNamespacedPodAsync(_podName, kubeNamespace);
 
-        var labels = new Dictionary<string, string>(pod.Metadata.Labels);
-
-        foreach (var kind in _kinds)
-        {
-            try
-            {
-                var labelKey = $"{LabelKind}-{kind}";
-                labels.Remove(labelKey);
-            }
-            catch (Exception x)
-            {
-                Logger.LogError(x, "[Cluster][KubernetesProvider] Failed to remove label");
-            }
-        }
+        var labels = pod.Metadata.Labels
+            .Where(label => !label.Key.StartsWith(ProtoClusterPrefix, StringComparison.Ordinal))
+            .ToDictionary(label => label.Key, label => label.Value);
 
-        labels.Remove(LabelCluster);
+        var annotations = pod.Metadata.Annotations
+            .Where(label => !label.Key.StartsWith(ProtoClusterPrefix, StringComparison.Ordinal))
+            .ToDictionary(label => label.Key, label => label.Value);
 
-        await kubernetes.ReplacePodLabels(_podName, kubeNamespace, pod, labels);
+        await kubernetes.ReplacePodLabelsAndAnnotations(_podName, kubeNamespace, pod, labels, annotations);
 
         cluster.System.Root.Send(_clusterMonitor, new DeregisterMember());
     }

src/Proto.Cluster.Kubernetes/ProtoLabels.cs

@@ -8,10 +8,9 @@ namespace Proto.Cluster.Kubernetes;
 
 public static class ProtoLabels
 {
-    private const string LabelPrefix = "cluster.proto.actor/";
-    public const string LabelPort = LabelPrefix + "port";
-    public const string LabelKind = LabelPrefix + "kind";
-    public const string LabelCluster = LabelPrefix + "cluster";
-    public const string LabelStatusValue = LabelPrefix + "status-value";
-    public const string LabelMemberId = LabelPrefix + "member-id";
+    public const string ProtoClusterPrefix = "cluster.proto.actor/";
+    public const string LabelPort = ProtoClusterPrefix + "port";
+    public const string LabelCluster = ProtoClusterPrefix + "cluster";
+    public const string LabelMemberId = ProtoClusterPrefix + "member-id";
+    public const string AnnotationKinds = ProtoClusterPrefix + "kinds";
 }