INFRA-2404 | Abhishek | Add separate serviceaccount and rolebinding for each flink cluster

templates/flink_role_binding.jsonnet

@@ -0,0 +1,25 @@
+local common = import 'common.jsonnet';
+local deployment_manifest = import 'deployment_manifest.jsonnet';
+local namespace = deployment_manifest.flink.namespace;
+
+{
+  apiVersion: 'rbac.authorization.k8s.io/v1',
+  kind: 'RoleBinding',
+  metadata: {
+    name: 'flink' + '-' + deployment_manifest.name + '-' + deployment_manifest.environment,
+    namespace: namespace,
+    labels: common.labels
+  },
+  roleRef: {
+    apiGroup: 'rbac.authorization.k8s.io',
+    kind: 'Role',
+    name: 'flink',
+  },
+  subjects: [
+    {
+      kind: "ServiceAccount",
+      name: deployment_manifest.name + '-' + deployment_manifest.environment,
+      namespace: namespace
+    }
+  ]
+}

templates/flink_service_account.jsonnet

@@ -0,0 +1,13 @@
+local common = import 'common.jsonnet';
+local deployment_manifest = import 'deployment_manifest.jsonnet';
+local namespace = deployment_manifest.flink.namespace;
+
+{
+  apiVersion: 'v1',
+  kind: 'ServiceAccount',
+  metadata: {
+    name: deployment_manifest.name + '-' + deployment_manifest.environment,
+    namespace: namespace,
+    labels: common.labels
+  }
+}

templates/main.jsonnet

@@ -36,6 +36,8 @@ local util = import 'util.jsonnet';
 local isSandbox = util.is_sandbox(deployment_manifest.environment);
 local flink_deployment = import 'flink_deployment.jsonnet';
 local flink_session_job = import 'flink_session_job.jsonnet';
+local flink_service_account = import 'flink_service_account.jsonnet';
+local flink_role_binding = import 'flink_role_binding.jsonnet';
 local isflinkJob = std.objectHas(deployment_manifest, 'flink');
 
 local flink_objects = (if isflinkJob then {
@@ -49,6 +51,8 @@ if isflinkJob then
      '0_secret.json': secret,
      '0_0_flink_deployment.json': flink_deployment,
      '0_1_flink_session_job.json': flink_session_job,
+     '0_2_flink_service_account.json': flink_service_account,
+     '0_3_flink_role_binding.json': flink_role_binding,
    } + { ['5_%s_ingress.json' % index]: ingresses[index] for index in std.range(0, std.length(ingresses) - 1) })
 else ({
   '0_secret.json': secret,