TelpoKafkaConsole/TelpoKafkaConsole.Service/KafkaAdminService.cs

using Confluent.Kafka;
using Confluent.Kafka.Admin;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Options;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
using TelpoKafkaConsole.Model;
using static System.Net.Mime.MediaTypeNames;

namespace TelpoKafkaConsole.Service
{
    public class KafkaAdminService
    {
        private readonly ILogger<KafkaAdminService> _logger;
        private readonly ServiceConfig _configService;
        public  IAdminClient _adminClient;
        public KafkaAdminService(ILogger<KafkaAdminService> logger, IOptions<ServiceConfig> _optConfigService) 
        { 
            _logger = logger;
            _configService = _optConfigService.Value;
            _adminClient = new AdminClientBuilder(new AdminClientConfig
            {
                BootstrapServers = _configService.KafkaServerAddress,
                SecurityProtocol = SecurityProtocol.SaslSsl,
                SaslMechanism = SaslMechanism.ScramSha256,
                SaslUsername = "superuser",
                SaslPassword = "password",
                SslCaLocation = _configService.KafkaServerPEMLocation
                // Add any other configuration options as needed
            }).Build();
        }

        public List<GroupInfo> ListGroups()
        {

            try
            {
                var groups = _adminClient.ListGroups(TimeSpan.FromSeconds(10));
                return groups;
            }
            catch (Exception ex)
            {
                throw new Exception(ex.Message);
            }
        }

        #region  UserScramCredentials

        public async Task<List<UserScramCredentialsDescription>> DescribeUserScramCredentialsAsync(IEnumerable<string> users)
        {
            try
            {

                var timeout = TimeSpan.FromSeconds(10);
                var descResult = await _adminClient.DescribeUserScramCredentialsAsync(users, new DescribeUserScramCredentialsOptions() { RequestTimeout = timeout });
                return descResult.UserScramCredentialsDescriptions;
                //foreach (var description in descResult.UserScramCredentialsDescriptions)
                //{
                //    Console.WriteLine($"  User: {description.User}");
                //    foreach (var scramCredentialInfo in description.ScramCredentialInfos)
                //    {
                //        Console.WriteLine($"    Mechanism: {scramCredentialInfo.Mechanism}");
                //        Console.WriteLine($"      Iterations: {scramCredentialInfo.Iterations}");
                //    }
                //}
            }
            catch (DescribeUserScramCredentialsException e)
            {
                if (e.Error.Code.ToString().Equals("Local_Partial")
                    && e.Results.UserScramCredentialsDescriptions.Count == 1
                    && e.Results.UserScramCredentialsDescriptions.First().ScramCredentialInfos.Count == 0
                    )
                {
                    return new List<UserScramCredentialsDescription>();
                }
                else
                {
                    var errMsg = $"An error occurred describing user SCRAM credentials for some users:\n";

                    foreach (var description in e.Results.UserScramCredentialsDescriptions)
                    {
                        errMsg += ($"User: {description.User} -- Error: {description.Error}\n");
                        if (!description.Error.IsError)
                        {
                            foreach (var scramCredentialInfo in description.ScramCredentialInfos)
                            {
                                errMsg += ($"Mechanism: {scramCredentialInfo.Mechanism} -- Iterations: {scramCredentialInfo.Iterations}\n");
                            }
                        }
                    }
                    throw new Exception(errMsg);
                }

               
            }
            catch (KafkaException e)
            {
               // _logger.LogError($"An error occurred describing user SCRAM credentials: {e}");
                throw new KafkaException(e.Error);

            }
            catch (Exception ex)
            {
                throw new Exception(ex.Message);
            }
        }

       
        public async Task AlterUserScramCredentialsAsync(ScramCredentialsUser scramUser, string Action = "UPSERT")
        {
            var alterations = new List<UserScramCredentialAlteration>();
            string user = scramUser.Name;
            var mechanism = scramUser.Mechanism;
            var iterations = scramUser.Iterations;
            var password = Encoding.UTF8.GetBytes(scramUser.Password);
            if (Action.Equals("DELETE"))
            {
                alterations.Add(new UserScramCredentialDeletion
                       {
                           User = user,
                           Mechanism = mechanism,
                       }
                   );
            }
            else
            {
                alterations.Add(new UserScramCredentialUpsertion
                  {
                      User = user,
                      ScramCredentialInfo = new ScramCredentialInfo
                      {
                          Mechanism = mechanism,
                          Iterations = iterations,
                      },
                      Password = password,
                      // Salt = salt,
                  }
              );
            }
            var timeout = TimeSpan.FromSeconds(30);
            try
            {
                await _adminClient.AlterUserScramCredentialsAsync(alterations,new AlterUserScramCredentialsOptions() { RequestTimeout = timeout });
                //_logger.LogError("All AlterUserScramCredentials operations completed successfully");
            }
            catch (AlterUserScramCredentialsException e)
            {
                var errMsg = ($"An error occurred altering user SCRAM credentials for some users:");
                foreach (var result in e.Results)
                {
                    errMsg += ($"User: {result.User} -- Error: {result.Error}\n");
                }
                throw new Exception( errMsg );  
            }
            catch (KafkaException e)
            {
                //_logger.LogError($"An error occurred altering user SCRAM credentials: {e}");
                throw new KafkaException( e.Error );
            }
            catch (Exception ex)
            {

                //_logger.LogError(ex.Message);           
                throw new Exception(ex.Message);
            }
        }

        #endregion

        #region ACLs
        public async Task<List<AclBinding>> DescribeAclsAsync()
        {
            //var name = "testtopic";//"";
            //var principal = "demo-consumer";
            var host = "*";
            List<AclBinding> ParseAclBindings = new()
            {
                new() {
                     Pattern = new ResourcePattern
                     {
                        Type = Confluent.Kafka.Admin.ResourceType.Any,//resourceType,
                        //Name = "demo-orders",
                        ResourcePatternType = ResourcePatternType.Any//resourcePatternType
                     },
                    Entry = new AccessControlEntry
                    {
                        //Principal ="User:demo-consumer",
                        Host = host,
                        Operation = AclOperation.Any,//operation,
                        PermissionType = AclPermissionType.Any//permissionType
                    }
                }
            };
            List<AclBindingFilter> aclBindingFilters = ParseAclBindings.Select(aclBinding => aclBinding.ToFilter()).ToList();
            try
            {
                var result = await _adminClient.DescribeAclsAsync(aclBindingFilters[0]);
                return result.AclBindings;
            }
            catch (DescribeAclsException e)
            {
                //_logger.LogError($"An error occurred in describe ACLs operation: Code: {e.Result.Error.Code}" +
                //    $", Reason: {e.Result.Error.Reason}");
                throw new Exception($"An error occurred in describe ACLs operation: Code: {e.Result.Error.Code}, Reason: {e.Result.Error.Reason}");
            }
            catch (KafkaException e)
            {
                throw new KafkaException(e.Error);
            }
            catch (Exception ex)
            {

                //_logger.LogError(ex.Message);
                throw new Exception(ex.Message);
            }

        }

        public async Task CreateAclsAsync(List<AclBinding> aclBindings)
        {
            try
            {
                await _adminClient.CreateAclsAsync(aclBindings);
                _logger.LogInformation("All create ACL operations completed successfully");
            }
            catch (CreateAclsException e)
            {
                var errMsg = ("One or more create ACL operations failed.\n");
                for (int i = 0; i < e.Results.Count; ++i)
                {
                    var result = e.Results[i];
                    if (!result.Error.IsError)
                    {
                        errMsg += ($"Create ACLs operation {i} completed successfully\n");
                    }
                    else
                    {
                        errMsg += ($"An error occurred in create ACL operation {i}: Code: {result.Error.Code}" +
                        $", Reason: {result.Error.Reason}\n");
                    }
                }
               
                throw new Exception(errMsg);
            }
            catch (KafkaException e)
            {
                //_logger.LogError($"An error occurred calling the CreateAcls operation: {e.Message}");
                throw new Exception($"An error occurred calling the CreateAcls operation: {e.Message}");
            }
            catch (Exception ex)
            { 

                //_logger.LogError(ex.Message);           
                throw new Exception(ex.Message);
            }
           
        }

        public async Task DeleteAclsAsync(List<AclBinding> aclBindings)
        {

            List<AclBindingFilter> aclBindingFilters = aclBindings.Select(aclBinding => aclBinding.ToFilter()).ToList();
            try
            {
                var results = await _adminClient.DeleteAclsAsync(aclBindingFilters);
                int i = 0;
                foreach (var result in results)
                {
                    _logger.LogInformation($"Deleted ACLs in operation {i}");
                   // PrintAclBindings(result.AclBindings);
                    ++i;
                }
            }
            catch (DeleteAclsException e)
            {
                var errMsg = ("One or more create ACL operations failed.\n");
                for (int i = 0; i < e.Results.Count; ++i)
                {
                    var result = e.Results[i];
                    if (!result.Error.IsError)
                    {
                        errMsg += ($"Deleted ACLs in operation {i}\n");
                       // PrintAclBindings(result.AclBindings);
                    }
                    else
                    {
                        errMsg += ($"An error occurred in delete ACL operation {i}: Code: {result.Error.Code}" +
                            $", Reason: {result.Error.Reason}\n");
                    }
                }

                throw new Exception (errMsg);
            }
            catch (KafkaException e)
            {
                throw new KafkaException(e.Error);
            }
            catch (Exception ex)
            {

                //_logger.LogError(ex.Message);           
                throw new Exception(ex.Message);
            }


        }
        #endregion

        #region Topics
        public async Task CreateTopic(string topicName, TimeSpan retentionTime,int numPartitions=1)
        {
            try
            {
                var configEntries = new Dictionary<string, string>
                {
                    { "retention.ms", ((int)retentionTime.TotalMilliseconds).ToString() }
                };

                await _adminClient.CreateTopicsAsync(
                    new TopicSpecification[] {
                        new() { 
                            Name = topicName, 
                            ReplicationFactor = 1, 
                            NumPartitions = numPartitions>3 ? 3 : numPartitions, 
                            Configs = configEntries 
                        } 
                });
            }
            catch (CreateTopicsException e)
            {
                throw new Exception($"An error occurred creating topic {e.Results[0].Topic}: {e.Results[0].Error.Reason}");
            }
        }
        public async Task DeleteTopics(IEnumerable<string> topicNames)
        {
            try
            {
                await _adminClient.DeleteTopicsAsync(topicNames);
            }
            catch (DeleteTopicsException e)
            {
                throw new Exception($"An error occurred deleting topic {e.Results[0].Topic}: {e.Results[0].Error.Reason}");
            }
        }
        public async Task<List<TopicDescription>> DescribeTopicsAsync(IEnumerable<string> topicNames)
        {
            try
            {
                // var topicCollection = TopicCollection.OfTopicNames(new[] { topicName });
                var topicCollection = TopicCollection.OfTopicNames(topicNames);
                var topicDescriptions = await _adminClient.DescribeTopicsAsync(topicCollection);
                return topicDescriptions.TopicDescriptions;
                //foreach (var topicDescription in topicDescriptions.TopicDescriptions)
                //{
                //    Console.WriteLine($"Topic: {topicDescription.Name}");
                //    foreach (var partition in topicDescription.Partitions)
                //    {
                //        Console.WriteLine($"Partition: {partition.Partition}, Leader: {partition.Leader}, Replicas: {string.Join(",", partition.Replicas)}, Isr: {string.Join(",", partition.ISR)}");
                //    }
                //}
            }
            catch (DescribeTopicsException e)
            {
                if (e.Error.Code.ToString().Equals("Local_Partial")
                   && e.Results.TopicDescriptions.First().Error.Code.ToString().Equals("UnknownTopicOrPart")
                   && e.Results.TopicDescriptions.First().Name.Equals(topicNames.First())
                   )
                {
                    return new List<TopicDescription>();
                }
                else
                {
                    throw new Exception($"An error occurred describing topic {e.Message}");

                }
               // throw new Exception($"An error occurred describing topic {e.Message}");
            }
        }

        #endregion


    }
}