Merge pull request #1 from sookeke/BCPSDEMS-577

feature branch for BCPSDEMS-577

.gitignore

@@ -22,3 +22,4 @@ ehthumbs_vista.db
 
 # Folder config file
 [Dd]esktop.ini
+/.vs

backend/service.edt/Controllers/WeatherForecastController.cs

@@ -33,7 +33,7 @@ public async Task<IEnumerable<WeatherForecast>> Get()
             })
             .ToArray();
 
-            await this._kafkaProducer.ProduceAsync("Dems_AccessRequest", Guid.NewGuid().ToString(), f.FirstOrDefault());
+            await this._kafkaProducer.ProduceAsync("beer-events", Guid.NewGuid().ToString(), f.FirstOrDefault());
 
             return f;
         }

backend/service.edt/EdtServiceConfiguration.cs

@@ -47,6 +47,14 @@ public class KafkaClusterConfiguration
         public string BoostrapServers { get; set; } = string.Empty;
         public string ConsumerTopicName { get; set; } = string.Empty;
         public string ProducerTopicName { get; set; } = string.Empty;
+        public string SaslOauthbearerTokenEndpointUrl { get; set; } = string.Empty;
+        public string SaslOauthbearerProducerClientId { get; set; } = string.Empty;
+        public string SaslOauthbearerProducerClientSecret { get; set; } = string.Empty;
+        public string SaslOauthbearerConsumerClientId { get; set; } = string.Empty;
+        public string SaslOauthbearerConsumerClientSecret { get; set; } = string.Empty;
+        public string SslCaLocation { get; set; } = string.Empty;
+        public string SslCertificateLocation { get; set; } = string.Empty;
+        public string SslKeyLocation { get; set; } = string.Empty;
     }
     public class JustinParticipantClientConfiguration
     {

backend/service.edt/Kafka/ConsumerSetup.cs

@@ -19,34 +19,36 @@ public static IServiceCollection AddKafkaConsumer(this IServiceCollection servic
         var clientConfig = new ClientConfig()
         {
             BootstrapServers = config.KafkaCluster.BoostrapServers,
-            SaslMechanism = SaslMechanism.Plain,
+            SaslMechanism = SaslMechanism.OAuthBearer,
             SecurityProtocol = SecurityProtocol.SaslSsl,
-            SaslUsername = config.KafkaCluster.ClientId,
-            SaslPassword = config.KafkaCluster.ClientSecret,
+            SaslOauthbearerTokenEndpointUrl = config.KafkaCluster.SaslOauthbearerTokenEndpointUrl,
+            SaslOauthbearerMethod = SaslOauthbearerMethod.Oidc,
+            SaslOauthbearerScope = "oidc",
+            SslEndpointIdentificationAlgorithm = SslEndpointIdentificationAlgorithm.Https,
+            SslCaLocation = config.KafkaCluster.SslCaLocation,
+            //SslCertificateLocation = config.KafkaCluster.SslCertificateLocation,
+            //SslKeyLocation = config.KafkaCluster.SslKeyLocation
         };
-        var producerConfig = new ProducerConfig
+        var producerConfig = new ProducerConfig(clientConfig)
         {
-            BootstrapServers = config.KafkaCluster.BoostrapServers,
             Acks = Acks.All,
-            SaslMechanism = SaslMechanism.Plain,
-            SecurityProtocol = SecurityProtocol.SaslSsl,
-            SaslUsername = config.KafkaCluster.ClientId,
-            SaslPassword = config.KafkaCluster.ClientSecret,
+            SaslOauthbearerClientId = config.KafkaCluster.SaslOauthbearerProducerClientId,
+            SaslOauthbearerClientSecret = config.KafkaCluster.SaslOauthbearerProducerClientSecret,
             EnableIdempotence = true
         };
 
         var consumerConfig = new ConsumerConfig(clientConfig)
         {
-            GroupId = "Dems-Consumer-Group",
+            GroupId = "accessrequest-consumer-group",
             EnableAutoCommit = true,
             AutoOffsetReset = AutoOffsetReset.Earliest,
-            BootstrapServers = config.KafkaCluster.BoostrapServers,
+            SaslOauthbearerClientId = config.KafkaCluster.SaslOauthbearerConsumerClientId,
+            SaslOauthbearerClientSecret = config.KafkaCluster.SaslOauthbearerConsumerClientSecret,
             EnableAutoOffsetStore = false,
             AutoCommitIntervalMs = 4000,
-            SaslMechanism = SaslMechanism.Plain,
-            SecurityProtocol = SecurityProtocol.SaslSsl,
-            SaslUsername = config.KafkaCluster.ClientId,
-            SaslPassword = config.KafkaCluster.ClientSecret
+            BootstrapServers = config.KafkaCluster.BoostrapServers,
+            SaslMechanism = SaslMechanism.OAuthBearer,
+            SecurityProtocol = SecurityProtocol.SaslSsl
         };
         //var producerConfig = new ProducerConfig(clientConfig);
         services.AddSingleton(consumerConfig);

backend/service.edt/Kafka/KafkaConsumer.cs

@@ -20,6 +20,15 @@ public KafkaConsumer(ConsumerConfig config, IServiceScopeFactory serviceScopeFac
         //this.consumer = consumer;
         //this.topic = topic;
     }
+    /// <summary>
+    /// for production use of sasl/oauthbearer
+    /// implement authentication callbackhandler for token retrival and refresh
+    /// https://docs.confluent.io/platform/current/kafka/authentication_sasl/authentication_sasl_oauth.html#production-use-of-sasl-oauthbearer
+    /// https://techcommunity.microsoft.com/t5/fasttrack-for-azure/event-hub-kafka-endpoint-azure-ad-authentication-using-c/ba-p/2586185
+    /// https://github.com/Azure/azure-event-hubs-for-kafka/issues/97
+    /// https://github.com/confluentinc/confluent-kafka-dotnet/blob/master/test/Confluent.Kafka.IntegrationTests/Tests/OauthBearerToken_PublishConsume.cs
+    /// </summary>
+    /// <param name="config"></param>
 
     public async Task Consume(string topic, CancellationToken stoppingToken)
     {

backend/service.edt/ServiceEvents/KafkaProducer.cs

@@ -1,12 +1,23 @@
 namespace edt.service.ServiceEvents;
 
 using Confluent.Kafka;
+using edt.service.HttpClients;
 using edt.service.Kafka.Interfaces;
 using EdtService.Kafka;
+using IdentityModel.Client;
 
 public class KafkaProducer<TKey, TValue> : IDisposable, IKafkaProducer<TKey, TValue> where TValue : class
 {
     private readonly IProducer<TKey, TValue> producer;
+    /// <summary>
+    /// for production use of sasl/oauthbearer
+    /// implement authentication callbackhandler for token retrival
+    /// https://docs.confluent.io/platform/current/kafka/authentication_sasl/authentication_sasl_oauth.html#production-use-of-sasl-oauthbearer
+    /// https://techcommunity.microsoft.com/t5/fasttrack-for-azure/event-hub-kafka-endpoint-azure-ad-authentication-using-c/ba-p/2586185
+    /// https://github.com/Azure/azure-event-hubs-for-kafka/issues/97
+    /// https://github.com/confluentinc/confluent-kafka-dotnet/blob/master/test/Confluent.Kafka.IntegrationTests/Tests/OauthBearerToken_PublishConsume.cs
+    /// </summary>
+    /// <param name="config"></param>
     public KafkaProducer(ProducerConfig config) => this.producer = new ProducerBuilder<TKey, TValue>(config).SetValueSerializer(new KafkaSerializer<TValue>()).Build();
     public async Task ProduceAsync(string topic, TKey key, TValue value) => await this.producer.ProduceAsync(topic, new Message<TKey, TValue> { Key = key, Value = value });
     public void Dispose()
@@ -15,4 +26,27 @@ public void Dispose()
         this.producer.Dispose();
         GC.SuppressFinalize(this);
     }
+    /// <summary>
+    /// create a reusable method for get accesstoken and refreshhandler for kafka clients in production
+    /// </summary>
+    /// <param name="producer"></param>
+    /// <param name="config"></param>
+    private async void TokenRefreshHandler(IProducer<TKey, TValue> producer, string config)
+    {
+        try
+        {
+            var accessTokenClient = new HttpClient();
+            var accessToken = await accessTokenClient.RequestClientCredentialsTokenAsync(new ClientCredentialsTokenRequest
+            {
+                Address = "https://sso-dev-5b7aa5-dev.apps.silver.devops.gov.bc.ca/auth/realms/DEMSPOC/protocol/openid-connect/token",
+                ClientId = "",
+                ClientSecret = "",
+            });
+            producer.OAuthBearerSetToken(accessToken.AccessToken, accessToken.ExpiresIn, null);
+        }
+        catch (Exception ex)
+        {
+            producer.OAuthBearerSetTokenFailure(ex.ToString());
+        }
+    }
 }

backend/service.edt/appsettings.json

@@ -14,9 +14,17 @@
     "ApiKey": "<redacted>"
   },
   "KafkaCluster": {
-    "BoostrapServers": "dems-kafka-ccobnjnq-o-c---gs--g.bf2.kafka.rhcloud.com:443",
-    "ConsumerTopicName": "Dems_AccessRequest",
-    "ProducerTopicName": "Dems_Notification",
+    "BoostrapServers": "dems-cluster-5b7aa5-dev.apps.silver.devops.gov.bc.ca:443",
+    "ConsumerTopicName": "dems-access-request",
+    "ProducerTopicName": "dem-notification-ack",
+    "SaslOauthbearerTokenEndpointUrl": "https://sso-dev-5b7aa5-dev.apps.silver.devops.gov.bc.ca/auth/realms/DEMSPOC/protocol/openid-connect/token",
+    "SaslOauthbearerProducerClientId": "kafka-producer",
+    "SaslOauthbearerProducerClientSecret": "<redacted>",
+    "SaslOauthbearerConsumerClientId": "kafka-consumer",
+    "SaslOauthbearerConsumerClientSecret": "<redacted>",
+    "SslCaLocation": "\\cert-manager\\truststore.cer.pem", //truststore CA for kafka cluster + keycloak CA auth server in pem
+    "SslCertificateLocation": "\\cert-manager\\ca.crt",
+    "SslKeyLocation": "\\cert-manager\\ca.key",
     "ClientId": "<redacted>",
     "ClientSecret": "<redacted>"
   },

backend/webapi/appsettings.json

@@ -35,7 +35,7 @@
     "Url": "https://localhost:7215/api"
   },
   "KafkaCluster": {
-    "BoostrapServers": "dems-kafka-ccobnjnq-o-c---gs--g.bf2.kafka.rhcloud.com:443",
+    "BoostrapServers": "dems-kafka-cd--b-mj--cba-h----a.bf2.kafka.rhcloud.com:443",
     "ProducerTopicName": "Dems_AccessRequest",
     "ConsumerTopicName": "Dems_Notification_Ack",
     "ClientId": "<redacted>",

charts/edt-service/templates/deployment.yaml

@@ -91,6 +91,11 @@ spec:
                 secretKeyRef:
                   name: kafkaconfig-edt
                   key: KafkaCluster__ClientSecret
+            - name: KafkaCluster__SslCaLocation
+              valueFrom:
+                 secretKeyRef:
+                   name: kafka-client-truststore
+                   key: kafka-client-truststore.cer.pem
             - name: ApplicationUrl
               value: "https://{{ if $isProd }}{{else}}{{ $release }}.{{end}}{{ $domain }}"
             - name: EdtCLient__Url

charts/edt-service/values.yaml

@@ -23,6 +23,8 @@ imagePullSecrets: []
 nameOverride: ""
 fullnameOverride: ""
 
+trustStoreCertificateSecretName: "kafka-client-truststore"
+
 database:
   # .NET Core database connection string
   dbConnectionString: 'host=postgresql;port=5432;database=pidpdb;username=postgres;password=postgres'

charts/integration/.helmignore

@@ -0,0 +1,23 @@
+# Patterns to ignore when building packages.
+# This supports shell glob matching, relative path matching, and
+# negation (prefixed with !). Only one pattern per line.
+.DS_Store
+# Common VCS dirs
+.git/
+.gitignore
+.bzr/
+.bzrignore
+.hg/
+.hgignore
+.svn/
+# Common backup files
+*.swp
+*.bak
+*.tmp
+*.orig
+*~
+# Various IDEs
+.project
+.idea/
+*.tmproj
+.vscode/

charts/integration/Chart.yaml

@@ -0,0 +1,24 @@
+apiVersion: v2
+name: integration
+description: A Helm chart for Kubernetes
+
+# A chart can be either an 'application' or a 'library' chart.
+#
+# Application charts are a collection of templates that can be packaged into versioned archives
+# to be deployed.
+#
+# Library charts provide useful utilities or functions for the chart developer. They're included as
+# a dependency of application charts to inject those utilities and functions into the rendering
+# pipeline. Library charts do not define any templates and therefore cannot be deployed.
+type: application
+
+# This is the chart version. This version number should be incremented each time you make changes
+# to the chart and its templates, including the app version.
+# Versions are expected to follow Semantic Versioning (https://semver.org/)
+version: 0.1.0
+
+# This is the version number of the application being deployed. This version number should be
+# incremented each time you make changes to the application. Versions are not expected to
+# follow Semantic Versioning. They should reflect the version the application is using.
+# It is recommended to use it with quotes.
+appVersion: "1.16.0"

charts/integration/templates/NOTES.txt

@@ -0,0 +1,22 @@
+1. Get the application URL by running these commands:
+{{- if .Values.ingress.enabled }}
+{{- range $host := .Values.ingress.hosts }}
+  {{- range .paths }}
+  http{{ if $.Values.ingress.tls }}s{{ end }}://{{ $host.host }}{{ .path }}
+  {{- end }}
+{{- end }}
+{{- else if contains "NodePort" .Values.service.type }}
+  export NODE_PORT=$(kubectl get --namespace {{ .Release.Namespace }} -o jsonpath="{.spec.ports[0].nodePort}" services {{ include "integration.fullname" . }})
+  export NODE_IP=$(kubectl get nodes --namespace {{ .Release.Namespace }} -o jsonpath="{.items[0].status.addresses[0].address}")
+  echo http://$NODE_IP:$NODE_PORT
+{{- else if contains "LoadBalancer" .Values.service.type }}
+     NOTE: It may take a few minutes for the LoadBalancer IP to be available.
+           You can watch the status of by running 'kubectl get --namespace {{ .Release.Namespace }} svc -w {{ include "integration.fullname" . }}'
+  export SERVICE_IP=$(kubectl get svc --namespace {{ .Release.Namespace }} {{ include "integration.fullname" . }} --template "{{"{{ range (index .status.loadBalancer.ingress 0) }}{{.}}{{ end }}"}}")
+  echo http://$SERVICE_IP:{{ .Values.service.port }}
+{{- else if contains "ClusterIP" .Values.service.type }}
+  export POD_NAME=$(kubectl get pods --namespace {{ .Release.Namespace }} -l "app.kubernetes.io/name={{ include "integration.name" . }},app.kubernetes.io/instance={{ .Release.Name }}" -o jsonpath="{.items[0].metadata.name}")
+  export CONTAINER_PORT=$(kubectl get pod --namespace {{ .Release.Namespace }} $POD_NAME -o jsonpath="{.spec.containers[0].ports[0].containerPort}")
+  echo "Visit http://127.0.0.1:8080 to use your application"
+  kubectl --namespace {{ .Release.Namespace }} port-forward $POD_NAME 8080:$CONTAINER_PORT
+{{- end }}

charts/integration/templates/_helpers.tpl

@@ -0,0 +1,62 @@
+{{/*
+Expand the name of the chart.
+*/}}
+{{- define "integration.name" -}}
+{{- default .Chart.Name .Values.nameOverride | trunc 63 | trimSuffix "-" }}
+{{- end }}
+
+{{/*
+Create a default fully qualified app name.
+We truncate at 63 chars because some Kubernetes name fields are limited to this (by the DNS naming spec).
+If release name contains chart name it will be used as a full name.
+*/}}
+{{- define "integration.fullname" -}}
+{{- if .Values.fullnameOverride }}
+{{- .Values.fullnameOverride | trunc 63 | trimSuffix "-" }}
+{{- else }}
+{{- $name := default .Chart.Name .Values.nameOverride }}
+{{- if contains $name .Release.Name }}
+{{- .Release.Name | trunc 63 | trimSuffix "-" }}
+{{- else }}
+{{- printf "%s-%s" .Release.Name $name | trunc 63 | trimSuffix "-" }}
+{{- end }}
+{{- end }}
+{{- end }}
+
+{{/*
+Create chart name and version as used by the chart label.
+*/}}
+{{- define "integration.chart" -}}
+{{- printf "%s-%s" .Chart.Name .Chart.Version | replace "+" "_" | trunc 63 | trimSuffix "-" }}
+{{- end }}
+
+{{/*
+Common labels
+*/}}
+{{- define "integration.labels" -}}
+helm.sh/chart: {{ include "integration.chart" . }}
+{{ include "integration.selectorLabels" . }}
+{{- if .Chart.AppVersion }}
+app.kubernetes.io/version: {{ .Chart.AppVersion | quote }}
+{{- end }}
+app.kubernetes.io/managed-by: {{ .Release.Service }}
+{{- end }}
+
+{{/*
+Selector labels
+*/}}
+{{- define "integration.selectorLabels" -}}
+app.kubernetes.io/name: {{ include "integration.name" . }}
+app.kubernetes.io/instance: {{ .Release.Name }}
+{{- end }}
+
+{{/*
+Create the name of the service account to use
+*/}}
+{{- define "integration.serviceAccountName" -}}
+{{- if .Values.serviceAccount.create }}
+{{- default (include "integration.fullname" .) .Values.serviceAccount.name }}
+{{- else }}
+{{- default "default" .Values.serviceAccount.name }}
+{{- end }}
+{{- end }}

charts/integration/templates/deployment.yaml

@@ -0,0 +1,61 @@
+apiVersion: apps/v1
+kind: Deployment
+metadata:
+  name: {{ include "integration.fullname" . }}
+  labels:
+    {{- include "integration.labels" . | nindent 4 }}
+spec:
+  {{- if not .Values.autoscaling.enabled }}
+  replicas: {{ .Values.replicaCount }}
+  {{- end }}
+  selector:
+    matchLabels:
+      {{- include "integration.selectorLabels" . | nindent 6 }}
+  template:
+    metadata:
+      {{- with .Values.podAnnotations }}
+      annotations:
+        {{- toYaml . | nindent 8 }}
+      {{- end }}
+      labels:
+        {{- include "integration.selectorLabels" . | nindent 8 }}
+    spec:
+      {{- with .Values.imagePullSecrets }}
+      imagePullSecrets:
+        {{- toYaml . | nindent 8 }}
+      {{- end }}
+      serviceAccountName: {{ include "integration.serviceAccountName" . }}
+      securityContext:
+        {{- toYaml .Values.podSecurityContext | nindent 8 }}
+      containers:
+        - name: {{ .Chart.Name }}
+          securityContext:
+            {{- toYaml .Values.securityContext | nindent 12 }}
+          image: "{{ .Values.image.repository }}:{{ .Values.image.tag | default .Chart.AppVersion }}"
+          imagePullPolicy: {{ .Values.image.pullPolicy }}
+          ports:
+            - name: http
+              containerPort: 80
+              protocol: TCP
+          livenessProbe:
+            httpGet:
+              path: /
+              port: http
+          readinessProbe:
+            httpGet:
+              path: /
+              port: http
+          resources:
+            {{- toYaml .Values.resources | nindent 12 }}
+      {{- with .Values.nodeSelector }}
+      nodeSelector:
+        {{- toYaml . | nindent 8 }}
+      {{- end }}
+      {{- with .Values.affinity }}
+      affinity:
+        {{- toYaml . | nindent 8 }}
+      {{- end }}
+      {{- with .Values.tolerations }}
+      tolerations:
+        {{- toYaml . | nindent 8 }}
+      {{- end }}