diff --git a/chart/templates/gatekeeper.yaml b/chart/templates/gatekeeper.yaml index 7247f39..925b7fb 100644 --- a/chart/templates/gatekeeper.yaml +++ b/chart/templates/gatekeeper.yaml @@ -46,8 +46,8 @@ spec: value: "{{ .Values.mattermost.verificationToken }}" - name: MATTERMOST_DOMAIN value: "{{ .Values.mattermost.domain }}" - - name: ORCHESTRATION_HOST - value: orchestration.{{ .Release.Namespace }}.svc.clusterset.local + - name: QUEUER_HOST + value: queuer.{{ .Release.Namespace }}.svc.clusterset.local resources: {{ toYaml .Values.gatekeeper.resources | nindent 10 }} restartPolicy: Always --- diff --git a/chart/templates/orchestration.yaml b/chart/templates/orchestration.yaml index b485e4c..775b9f0 100644 --- a/chart/templates/orchestration.yaml +++ b/chart/templates/orchestration.yaml @@ -1,24 +1,4 @@ {{ if eq .Values.orchestration.enabled true }} -apiVersion: v1 -kind: Service -metadata: - name: orchestration - labels: - app: orchestration -spec: - type: ClusterIP - ports: - - name: http - port: 80 - targetPort: 80 - selector: - app: orchestration ---- -kind: ServiceExport -apiVersion: net.gke.io/v1 -metadata: - name: orchestration ---- apiVersion: apps/v1 kind: Deployment metadata: @@ -37,12 +17,14 @@ spec: sidecar.istio.io/inject: "true" istio.io/rev: "asm-managed-rapid" spec: - serviceAccountName: db-client-microservice + serviceAccountName: orchestration-microservice containers: - name: orchestration image: {{ .Values.orchestration.image.repository }}:{{ .Values.orchestration.image.tag }} imagePullPolicy: Always env: + - name: PUBSUB_PROJECT + value: {{ .Values.project }} - name: SPANNER_INSTANCE_ID value: "isidro" - name: SPANNER_DATABASE_ID @@ -68,19 +50,6 @@ spec: resources: {{ toYaml .Values.orchestration.resources | nindent 10 }} restartPolicy: Always --- -apiVersion: monitoring.googleapis.com/v1alpha1 -kind: PodMonitoring -metadata: - name: orchestration -spec: - selector: - matchLabels: - app: orchestration - endpoints: - - port: 80 - interval: 30s - timeout: 10s ---- apiVersion: networking.k8s.io/v1 kind: NetworkPolicy metadata: @@ -92,27 +61,6 @@ spec: policyTypes: - Egress - Ingress - ingress: - - from: - # GCP Health Check IPs - - ipBlock: - cidr: 35.191.0.0/16 - - ipBlock: - cidr: 130.211.0.0/22 - # RFC1918 (overkill, but enables multi-cluster and multi-region) - - ipBlock: - cidr: 10.0.0.0/8 - - ipBlock: - cidr: 172.16.0.0/12 - - ipBlock: - cidr: 192.168.0.0/16 - # In-cluster services (failover for non-RFC1918 topologies) - - podSelector: - matchLabels: - app: gatekeeper - - namespaceSelector: - matchLabels: - kubernetes.io/metadata.name: gmp-system egress: - {} {{ end }} \ No newline at end of file diff --git a/chart/templates/queuer.yaml b/chart/templates/queuer.yaml new file mode 100644 index 0000000..305e777 --- /dev/null +++ b/chart/templates/queuer.yaml @@ -0,0 +1,98 @@ +{{ if eq .Values.queuer.enabled true }} +apiVersion: v1 +kind: Service +metadata: + name: queuer + labels: + app: queuer +spec: + type: ClusterIP + ports: + - name: http + port: 80 + targetPort: 80 + selector: + app: queuer +--- +kind: ServiceExport +apiVersion: net.gke.io/v1 +metadata: + name: queuer +--- +apiVersion: apps/v1 +kind: Deployment +metadata: + name: queuer + labels: + app: queuer +spec: + replicas: 1 + selector: + matchLabels: + app: queuer + template: + metadata: + labels: + app: queuer + sidecar.istio.io/inject: "true" + istio.io/rev: "asm-managed-rapid" + spec: + serviceAccountName: orchestration-microservice + containers: + - name: queuer + image: {{ .Values.queuer.image.repository }}:{{ .Values.queuer.image.tag }} + imagePullPolicy: Always + env: + - name: PUBSUB_PROJECT + value: {{ .Values.project }} + resources: {{ toYaml .Values.queuer.resources | nindent 10 }} + restartPolicy: Always +--- +apiVersion: monitoring.googleapis.com/v1alpha1 +kind: PodMonitoring +metadata: + name: queuer +spec: + selector: + matchLabels: + app: queuer + endpoints: + - port: 80 + interval: 30s + timeout: 10s +--- +apiVersion: networking.k8s.io/v1 +kind: NetworkPolicy +metadata: + name: queuer +spec: + podSelector: + matchLabels: + app: queuer + policyTypes: + - Egress + - Ingress + ingress: + - from: + # GCP Health Check IPs + - ipBlock: + cidr: 35.191.0.0/16 + - ipBlock: + cidr: 130.211.0.0/22 + # RFC1918 (overkill, but enables multi-cluster and multi-region) + - ipBlock: + cidr: 10.0.0.0/8 + - ipBlock: + cidr: 172.16.0.0/12 + - ipBlock: + cidr: 192.168.0.0/16 + # In-cluster services (failover for non-RFC1918 topologies) + - podSelector: + matchLabels: + app: gatekeeper + - namespaceSelector: + matchLabels: + kubernetes.io/metadata.name: gmp-system + egress: + - {} +{{ end }} \ No newline at end of file diff --git a/chart/templates/service-account.yaml b/chart/templates/service-account.yaml index 4384494..c7f427d 100644 --- a/chart/templates/service-account.yaml +++ b/chart/templates/service-account.yaml @@ -9,9 +9,9 @@ metadata: apiVersion: v1 kind: ServiceAccount metadata: - name: db-client-microservice + name: orchestration-microservice annotations: - iam.gke.io/gcp-service-account: isidro-db-client-microservices@{{ .Values.project }}.iam.gserviceaccount.com + iam.gke.io/gcp-service-account: isidro-orchestration@{{ .Values.project }}.iam.gserviceaccount.com --- apiVersion: v1 kind: ServiceAccount diff --git a/chart/values.yaml b/chart/values.yaml index bdd6224..3dfa2ed 100644 --- a/chart/values.yaml +++ b/chart/values.yaml @@ -251,6 +251,19 @@ policyAgent: enabled: false config: {} +queuer: + enabled: true + image: + repository: us.gcr.io/PROJECT/isidro/queuer + tag: latest + resources: + requests: + cpu: 250m + memory: 1Gi + limits: + cpu: 1 + memory: 1Gi + repeater: enabled: true image: diff --git a/gatekeeper/main.py b/gatekeeper/main.py index 18d6e64..b4d7550 100644 --- a/gatekeeper/main.py +++ b/gatekeeper/main.py @@ -8,7 +8,7 @@ MATTERMOST_ACCESS_TOKEN = os.environ.get("MATTERMOST_ACCESS_TOKEN") MATTERMOST_VERIFICATION_TOKEN = os.environ.get("MATTERMOST_VERIFICATION_TOKEN") MATTERMOST_DOMAIN = os.environ.get("MATTERMOST_DOMAIN") -ORCHESTRATION_HOST = os.environ.get("ORCHESTRATION_HOST") +QUEUER_HOST = os.environ.get("QUEUER_HOST") if not SLACK_VERIFICATION_TOKEN: raise ValueError("No SLACK_VERIFICATION_TOKEN environment variable set") @@ -18,8 +18,8 @@ raise ValueError("No MATTERMOST_VERIFICATION_TOKEN environment variable set") if not MATTERMOST_DOMAIN: raise ValueError("No MATTERMOST_DOMAIN environment variable set") -if not ORCHESTRATION_HOST: - raise ValueError("No ORCHESTRATION_HOST environment variable set") +if not QUEUER_HOST: + raise ValueError("No QUEUER_HOST environment variable set") app = Flask(__name__) @@ -127,7 +127,7 @@ def challenge_response(self): def general_response(self): # Pass data to the orchestration service requests.post( - f"http://{ORCHESTRATION_HOST}/v1/orchestrate", + f"http://{QUEUER_HOST}/v1/queue", json={ "platform": self.get_platform(), "channel": self.get_channel(), diff --git a/orchestration/Dockerfile b/orchestration/Dockerfile index 6f08a6e..843f9b1 100644 --- a/orchestration/Dockerfile +++ b/orchestration/Dockerfile @@ -13,4 +13,4 @@ COPY . . ENV PYTHONUNBUFFERED True -CMD exec gunicorn --bind :80 --workers 1 --threads 8 --timeout 0 --log-level=debug main:app +CMD python ./main.py diff --git a/orchestration/main.py b/orchestration/main.py index af2c179..094eddb 100644 --- a/orchestration/main.py +++ b/orchestration/main.py @@ -3,12 +3,10 @@ import re import requests -import metrics import observability -from flask import Flask, abort, request +from google.cloud import pubsub_v1 from spanner import database, insert_post -from prometheus_client import generate_latest CONFIRMATION_WORDS = [ "yes", @@ -22,6 +20,7 @@ "sure", ] +PUBSUB_PROJECT = os.environ.get("PUBSUB_PROJECT") GREETING = os.environ.get("GREETING") RESPONDER_HOST = os.environ.get("RESPONDER_HOST") KEYWORDS_HOST = os.environ.get("KEYWORDS_HOST") @@ -32,6 +31,8 @@ TASKS_HOST = os.environ.get("TASKS_HOST") REPEATER_HOST = os.environ.get("REPEATER_HOST") +if not PUBSUB_PROJECT: + raise ValueError("No PUBSUB_PROJECT environment variable set") if not GREETING: raise ValueError("No GREETING environment variable set") if not RESPONDER_HOST: @@ -51,18 +52,19 @@ if not REPEATER_HOST: raise ValueError("No REPEATER_HOST environment variable set") -app = Flask(__name__) -trace = observability.setup(flask_app=app, requests_enabled=True) +trace = observability.setup(requests_enabled=True) db = database() +subscriber = pubsub_v1.SubscriberClient() + class Orchestration: - def __init__(self, request): - request = request.get_json() - self.platform = request["platform"] - self.channel = request["channel"] - self.thread_ts = request["thread_ts"] - self.user = request["user"] - self.text = request["text"] + def __init__(self, message): + message = json.loads(message.data) + self.platform = message["platform"] + self.channel = message["channel"] + self.thread_ts = message["thread_ts"] + self.user = message["user"] + self.text = message["text"] self.confirmed = False self.confirmation_text = None self.action = None @@ -296,15 +298,13 @@ def fail_interpolation(self): abort(400) -@app.route("/v1/orchestrate", methods=["POST"]) -def orchestrate(): - orchestration = Orchestration(request) +def callback(message): + orchestration = Orchestration(message) orchestration.insert_post() orchestration.confirmation() if not orchestration.confirmed: orchestration.get_action() orchestration.send_confirmation() - return "" elif orchestration.user_is_confirming(): orchestration.get_action() if orchestration.is_gbash_async(): @@ -320,19 +320,15 @@ def orchestrate(): # Fallback is to defer to async task system elif orchestration.is_async(): orchestration.send_task() - return "" elif not orchestration.user_is_confirming(): orchestration.send_rejection() - return "" - else: - abort(400) - + message.ack() -@app.route("/metrics") -def metrics(): - return generate_latest() +streaming_pull_future = subscriber.subscribe( + f"projects/{PUBSUB_PROJECT}/subscriptions/isidro-requests-orchestration", + callback=callback, +) -@app.route("/", methods=["GET"]) -def health(): - return "" +with subscriber: + streaming_pull_future.result() diff --git a/orchestration/metrics.py b/orchestration/metrics.py deleted file mode 100644 index bf2ff87..0000000 --- a/orchestration/metrics.py +++ /dev/null @@ -1,37 +0,0 @@ -from spanner import database -from prometheus_client import Gauge - -db = database() - -def channels(): - with db.snapshot() as snapshot: - result = snapshot.execute_sql(f"SELECT COUNT(DISTINCT channel) FROM posts") - return result.one()[0] - - -channels_metric = Gauge( - "isidro:channels", "Number of channels containing Isidro conversations" -) -channels_metric.set_function(channels) - - -def threads(): - with db.snapshot() as snapshot: - result = snapshot.execute_sql(f"SELECT COUNT(DISTINCT thread_ts) FROM posts") - return result.one()[0] - - -threads_metric = Gauge("isidro:threads", "Number of Isidro conversations") -threads_metric.set_function(threads) - - -def users(): - with db.snapshot() as snapshot: - result = snapshot.execute_sql(f"SELECT COUNT(DISTINCT user) FROM posts") - return result.one()[0] - - -users_metric = Gauge( - "isidro:users", "Number of users who have had Isidro conversations" -) -users_metric.set_function(users) diff --git a/orchestration/requirements-pypi.txt b/orchestration/requirements-pypi.txt index 9464ece..115be0b 100644 --- a/orchestration/requirements-pypi.txt +++ b/orchestration/requirements-pypi.txt @@ -1,5 +1,4 @@ flask>=2.0.2,<3.0.0 +google-cloud-pubsub>=2.11.0,<3.0.0 google-cloud-spanner>=3.15.1,<4.0.0 -gunicorn>=20.1.0,<21.0.0 -prometheus_client>=0.14.0,<1.0.0 requests>=2.26.0,<3.0.0 \ No newline at end of file diff --git a/provisioning/modules/foundation/dataset.tf b/provisioning/modules/foundation/dataset.tf new file mode 100644 index 0000000..0c7a0e9 --- /dev/null +++ b/provisioning/modules/foundation/dataset.tf @@ -0,0 +1,3 @@ +resource "google_bigquery_dataset" "isidro" { + dataset_id = "isidro" +} \ No newline at end of file diff --git a/provisioning/modules/foundation/iam-microservices.tf b/provisioning/modules/foundation/iam-microservices.tf index b11de11..0ec8aef 100644 --- a/provisioning/modules/foundation/iam-microservices.tf +++ b/provisioning/modules/foundation/iam-microservices.tf @@ -15,27 +15,33 @@ resource "google_project_iam_member" "tracing_microservices_workload_identity_us member = "serviceAccount:${var.project_id}.svc.id.goog[isidro/tracing-microservice]" } -resource "google_service_account" "db_microservices" { - account_id = "isidro-db-client-microservices" - display_name = "Isidro DB Client Microservices" +resource "google_service_account" "orchestration_microservices" { + account_id = "isidro-orchestration" + display_name = "Isidro Orchestration Microservices" } -resource "google_project_iam_member" "db_microservices_spanner_user" { +resource "google_project_iam_member" "orchestration_microservices_spanner_user" { project = var.project_id role = "roles/spanner.databaseUser" - member = "serviceAccount:${google_service_account.db_microservices.email}" + member = "serviceAccount:${google_service_account.orchestration_microservices.email}" } -resource "google_project_iam_member" "db_microservices_trace_writer" { +resource "google_project_iam_member" "orchestration_microservices_trace_writer" { project = var.project_id role = "roles/cloudtrace.agent" - member = "serviceAccount:${google_service_account.db_microservices.email}" + member = "serviceAccount:${google_service_account.orchestration_microservices.email}" } -resource "google_project_iam_member" "db_microservices_workload_identity_user" { +resource "google_project_iam_member" "orchestration_microservices_pubsub_editor" { + project = var.project_id + role = "roles/pubsub.editor" + member = "serviceAccount:${google_service_account.orchestration_microservices.email}" +} + +resource "google_project_iam_member" "orchestration_microservices_workload_identity_user" { project = var.project_id role = "roles/iam.workloadIdentityUser" - member = "serviceAccount:${var.project_id}.svc.id.goog[isidro/db-client-microservice]" + member = "serviceAccount:${var.project_id}.svc.id.goog[isidro/orchestration-microservice]" } resource "google_service_account" "kubebash_microservices" { diff --git a/provisioning/modules/foundation/pubsub.tf b/provisioning/modules/foundation/pubsub.tf new file mode 100644 index 0000000..216fc59 --- /dev/null +++ b/provisioning/modules/foundation/pubsub.tf @@ -0,0 +1,138 @@ +resource "google_pubsub_topic" "requests" { + name = "isidro-requests" + # 24 hour message retention + message_retention_duration = "86600s" + depends_on = [google_pubsub_schema.requests] + schema_settings { + schema = "projects/${var.project_id}/schemas/isidro-requests" + encoding = "JSON" + } +} + +resource "google_pubsub_schema" "requests" { + name = "isidro-requests" + type = "AVRO" + definition = <=0.1.0,<1.0.0 \ No newline at end of file diff --git a/queuer/requirements-pypi.txt b/queuer/requirements-pypi.txt new file mode 100644 index 0000000..e6a5e36 --- /dev/null +++ b/queuer/requirements-pypi.txt @@ -0,0 +1,3 @@ +flask>=2.0.2,<3.0.0 +google-cloud-pubsub>=2.11.0,<3.0.0 +gunicorn>=20.1.0,<21.0.0 \ No newline at end of file diff --git a/roles/provisioner.yaml b/roles/provisioner.yaml index 9d33dea..26be317 100644 --- a/roles/provisioner.yaml +++ b/roles/provisioner.yaml @@ -9,6 +9,10 @@ includedPermissions: - artifactregistry.repositories.list - artifactregistry.repositories.setIamPolicy - artifactregistry.repositories.update +- bigquery.datasets.create +- bigquery.datasets.delete +- bigquery.datasets.get +- bigquery.datasets.update - binaryauthorization.attestors.create - binaryauthorization.attestors.delete - binaryauthorization.attestors.get @@ -103,6 +107,24 @@ includedPermissions: - monitoring.slos.create - monitoring.slos.delete - monitoring.slos.get +- pubsub.schemas.attach +- pubsub.schemas.create +- pubsub.schemas.delete +- pubsub.schemas.get +- pubsub.schemas.list +- pubsub.schemas.validate +- pubsub.subscriptions.create +- pubsub.subscriptions.delete +- pubsub.subscriptions.get +- pubsub.subscriptions.list +- pubsub.subscriptions.update +- pubsub.topics.attachSubscription +- pubsub.topics.create +- pubsub.topics.delete +- pubsub.topics.detachSubscription +- pubsub.topics.get +- pubsub.topics.list +- pubsub.topics.update - serviceusage.services.disable - serviceusage.services.enable - serviceusage.services.get diff --git a/skaffold.dev.yaml b/skaffold.dev.yaml index ba57b5a..e5b3049 100644 --- a/skaffold.dev.yaml +++ b/skaffold.dev.yaml @@ -49,6 +49,15 @@ build: hooks: after: - command: ["bash", "-c", "./skaffold-binauthz.sh dev $SKAFFOLD_IMAGE"] + - image: us.gcr.io/GOOGLE_PROJECT/isidro/queuer + context: queuer + kaniko: + cache: {} + buildArgs: + REGISTRY_PROJECT: GOOGLE_PROJECT + hooks: + after: + - command: ["bash", "-c", "./skaffold-binauthz.sh dev $SKAFFOLD_IMAGE"] - image: us.gcr.io/GOOGLE_PROJECT/isidro/repeater context: repeater kaniko: @@ -182,6 +191,7 @@ deploy: keywords.image: us.gcr.io/GOOGLE_PROJECT/isidro/keywords kubebash.image: us.gcr.io/GOOGLE_PROJECT/isidro/kubebash orchestration.image: us.gcr.io/GOOGLE_PROJECT/isidro/orchestration + queuer.image: us.gcr.io/GOOGLE_PROJECT/isidro/queuer repeater.image: us.gcr.io/GOOGLE_PROJECT/isidro/repeater responder.image: us.gcr.io/GOOGLE_PROJECT/isidro/responder tasks.image: us.gcr.io/GOOGLE_PROJECT/isidro/tasks diff --git a/skaffold.prod.yaml b/skaffold.prod.yaml index 2be92b7..9cb7cce 100644 --- a/skaffold.prod.yaml +++ b/skaffold.prod.yaml @@ -49,6 +49,15 @@ build: hooks: after: - command: ["bash", "-c", "./skaffold-binauthz.sh prod $SKAFFOLD_IMAGE"] + - image: us.gcr.io/GOOGLE_PROJECT/isidro/queuer + context: queuer + kaniko: + cache: {} + buildArgs: + REGISTRY_PROJECT: GOOGLE_PROJECT + hooks: + after: + - command: ["bash", "-c", "./skaffold-binauthz.sh dev $SKAFFOLD_IMAGE"] - image: us.gcr.io/GOOGLE_PROJECT/isidro/repeater context: repeater kaniko: @@ -182,6 +191,7 @@ deploy: keywords.image: us.gcr.io/GOOGLE_PROJECT/isidro/keywords kubebash.image: us.gcr.io/GOOGLE_PROJECT/isidro/kubebash orchestration.image: us.gcr.io/GOOGLE_PROJECT/isidro/orchestration + queuer.image: us.gcr.io/GOOGLE_PROJECT/isidro/queuer repeater.image: us.gcr.io/GOOGLE_PROJECT/isidro/repeater responder.image: us.gcr.io/GOOGLE_PROJECT/isidro/responder tasks.image: us.gcr.io/GOOGLE_PROJECT/isidro/tasks @@ -219,6 +229,7 @@ deploy: keywords.image: us.gcr.io/GOOGLE_PROJECT/isidro/keywords kubebash.image: us.gcr.io/GOOGLE_PROJECT/isidro/kubebash orchestration.image: us.gcr.io/GOOGLE_PROJECT/isidro/orchestration + queuer.image: us.gcr.io/GOOGLE_PROJECT/isidro/queuer repeater.image: us.gcr.io/GOOGLE_PROJECT/isidro/repeater responder.image: us.gcr.io/GOOGLE_PROJECT/isidro/responder tasks.image: us.gcr.io/GOOGLE_PROJECT/isidro/tasks @@ -256,6 +267,7 @@ deploy: keywords.image: us.gcr.io/GOOGLE_PROJECT/isidro/keywords kubebash.image: us.gcr.io/GOOGLE_PROJECT/isidro/kubebash orchestration.image: us.gcr.io/GOOGLE_PROJECT/isidro/orchestration + queuer.image: us.gcr.io/GOOGLE_PROJECT/isidro/queuer repeater.image: us.gcr.io/GOOGLE_PROJECT/isidro/repeater responder.image: us.gcr.io/GOOGLE_PROJECT/isidro/responder tasks.image: us.gcr.io/GOOGLE_PROJECT/isidro/tasks @@ -293,6 +305,7 @@ deploy: keywords.image: us.gcr.io/GOOGLE_PROJECT/isidro/keywords kubebash.image: us.gcr.io/GOOGLE_PROJECT/isidro/kubebash orchestration.image: us.gcr.io/GOOGLE_PROJECT/isidro/orchestration + queuer.image: us.gcr.io/GOOGLE_PROJECT/isidro/queuer repeater.image: us.gcr.io/GOOGLE_PROJECT/isidro/repeater responder.image: us.gcr.io/GOOGLE_PROJECT/isidro/responder tasks.image: us.gcr.io/GOOGLE_PROJECT/isidro/tasks @@ -330,6 +343,7 @@ deploy: keywords.image: us.gcr.io/GOOGLE_PROJECT/isidro/keywords kubebash.image: us.gcr.io/GOOGLE_PROJECT/isidro/kubebash orchestration.image: us.gcr.io/GOOGLE_PROJECT/isidro/orchestration + queuer.image: us.gcr.io/GOOGLE_PROJECT/isidro/queuer repeater.image: us.gcr.io/GOOGLE_PROJECT/isidro/repeater responder.image: us.gcr.io/GOOGLE_PROJECT/isidro/responder tasks.image: us.gcr.io/GOOGLE_PROJECT/isidro/tasks