From fbd565651a1aad708572dbceb888233d19b98c09 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E2=80=9CNaeel=E2=80=9D?= Date: Mon, 11 May 2026 09:37:20 +0400 Subject: [PATCH] feat(provider): add fission_mq_trigger, fission_cron_trigger, fission_iot_device + weather-demo example --- examples/weather-demo/consumer/main.py | 45 +++ .../weather-demo/consumer/requirements.txt | 1 + examples/weather-demo/fetcher/main.py | 68 +++++ .../weather-demo/fetcher/requirements.txt | 2 + examples/weather-demo/main.tf | 149 ++++++++++ examples/weather-demo/migration.sql | 18 ++ .../weather-demo/terraform.tfvars.example | 10 + terraform/provider/internal/client/client.go | 264 ++++++++++++++++++ .../provider/internal/provider/provider.go | 3 + .../resources/cron_trigger_resource.go | 208 ++++++++++++++ .../internal/resources/iot_device_resource.go | 230 +++++++++++++++ .../internal/resources/mq_trigger_resource.go | 205 ++++++++++++++ 12 files changed, 1203 insertions(+) create mode 100644 examples/weather-demo/consumer/main.py create mode 100644 examples/weather-demo/consumer/requirements.txt create mode 100644 examples/weather-demo/fetcher/main.py create mode 100644 examples/weather-demo/fetcher/requirements.txt create mode 100644 examples/weather-demo/main.tf create mode 100644 examples/weather-demo/migration.sql create mode 100644 examples/weather-demo/terraform.tfvars.example create mode 100644 terraform/provider/internal/resources/cron_trigger_resource.go create mode 100644 terraform/provider/internal/resources/iot_device_resource.go create mode 100644 terraform/provider/internal/resources/mq_trigger_resource.go diff --git a/examples/weather-demo/consumer/main.py b/examples/weather-demo/consumer/main.py new file mode 100644 index 0000000..56de1fe --- /dev/null +++ b/examples/weather-demo/consumer/main.py @@ -0,0 +1,45 @@ +import os +import json +import psycopg2 + + +def main(event, context): + pg_dsn = os.environ["PG_DSN"] + + # Извлекаем тело сообщения из SQS (POST от sqs-consumer) + body = getattr(event, "body", event) + if isinstance(body, (bytes, bytearray)): + body = body.decode("utf-8") + if isinstance(body, str): + data = json.loads(body) + else: + data = body + + conn = psycopg2.connect(pg_dsn) + try: + cur = conn.cursor() + cur.execute( + """ + INSERT INTO weather_metrics + (city, country, temperature, feels_like, humidity, + pressure, wind_speed, description, recorded_at) + VALUES (%s, %s, %s, %s, %s, %s, %s, %s, to_timestamp(%s)) + """, + ( + data["city"], + data["country"], + data["temperature"], + data["feels_like"], + data["humidity"], + data["pressure"], + data["wind_speed"], + data["description"], + data["owm_timestamp"], + ), + ) + conn.commit() + cur.close() + finally: + conn.close() + + return {"status": "ok", "city": data["city"], "temp": data["temperature"]} diff --git a/examples/weather-demo/consumer/requirements.txt b/examples/weather-demo/consumer/requirements.txt new file mode 100644 index 0000000..58ab769 --- /dev/null +++ b/examples/weather-demo/consumer/requirements.txt @@ -0,0 +1 @@ +psycopg2-binary==2.9.9 diff --git a/examples/weather-demo/fetcher/main.py b/examples/weather-demo/fetcher/main.py new file mode 100644 index 0000000..a2affc3 --- /dev/null +++ b/examples/weather-demo/fetcher/main.py @@ -0,0 +1,68 @@ +import os +import json +import requests +import boto3 +from botocore.config import Config + + +CITIES = [ + "Moscow,RU", + "London,GB", + "Paphos,CY", + "Ulyanovsk,RU", + "Santiago,CL", +] + + +def main(event, context): + api_key = os.environ["OWM_API_KEY"] + sqs_endpoint = os.environ.get( + "SQS_ENDPOINT", "http://shared-sqs.shared-sqs.svc.cluster.local:4100" + ) + access_key = os.environ["SQS_ACCESS_KEY"] + secret_key = os.environ["SQS_SECRET_KEY"] + queue_name = os.environ.get("SQS_QUEUE_NAME", "weather-data") + + cfg = Config(signature_version="s3v4", s3={"addressing_style": "path"}) + sqs = boto3.client( + "sqs", + endpoint_url=sqs_endpoint, + aws_access_key_id=access_key, + aws_secret_access_key=secret_key, + region_name="us-east-1", + config=cfg, + ) + + # Создаём очередь если не существует + try: + queue_url = sqs.get_queue_url(QueueName=queue_name)["QueueUrl"] + except Exception: + queue_url = sqs.create_queue(QueueName=queue_name)["QueueUrl"] + + results = [] + for city in CITIES: + try: + resp = requests.get( + "https://api.openweathermap.org/data/2.5/weather", + params={"q": city, "appid": api_key, "units": "metric"}, + timeout=10, + ) + resp.raise_for_status() + data = resp.json() + msg = { + "city": data["name"], + "country": data["sys"]["country"], + "temperature": data["main"]["temp"], + "feels_like": data["main"]["feels_like"], + "humidity": data["main"]["humidity"], + "pressure": data["main"]["pressure"], + "wind_speed": data["wind"]["speed"], + "description": data["weather"][0]["description"], + "owm_timestamp": data["dt"], + } + sqs.send_message(QueueUrl=queue_url, MessageBody=json.dumps(msg)) + results.append({"city": msg["city"], "temp": msg["temperature"]}) + except Exception as e: + results.append({"city": city, "error": str(e)}) + + return {"status": "ok", "sent": len(results), "results": results} diff --git a/examples/weather-demo/fetcher/requirements.txt b/examples/weather-demo/fetcher/requirements.txt new file mode 100644 index 0000000..aec8898 --- /dev/null +++ b/examples/weather-demo/fetcher/requirements.txt @@ -0,0 +1,2 @@ +requests==2.31.0 +boto3==1.34.0 diff --git a/examples/weather-demo/main.tf b/examples/weather-demo/main.tf new file mode 100644 index 0000000..1aab697 --- /dev/null +++ b/examples/weather-demo/main.tf @@ -0,0 +1,149 @@ +terraform { + required_providers { + fission = { + source = "nail/fission" + version = "~> 0.2.0" + } + } +} + +provider "fission" { + kubeconfig_path = var.kubeconfig_path + namespace = var.namespace +} + +# ── Переменные ─────────────────────────────────────────────────────── + +variable "kubeconfig_path" { + default = "/home/naeel/.kube/config" + description = "Путь к kubeconfig." +} + +variable "namespace" { + default = "fission-weather" + description = "Namespace для функций и триггеров." +} + +variable "owm_api_key" { + sensitive = true + description = "OpenWeatherMap API key." +} + +variable "sqs_access_key" { + sensitive = true + description = "SQS Access Key." +} + +variable "sqs_secret_key" { + sensitive = true + description = "SQS Secret Key." +} + +variable "pg_dsn" { + sensitive = true + description = "PostgreSQL DSN для записи метрик. Пример: postgresql://user:pass@host:5432/db?sslmode=disable" +} + +# ── IoT-устройство — виртуальная метеостанция ──────────────────────── + +resource "fission_iot_device" "weather_station" { + name = "weather-station" + device_id = "weather-station-01" + namespace = "sless" + metadata = { + type = "weather-station" + location = "multi-city" + cities = "Moscow,London,Paphos,Ulyanovsk,Santiago" + } +} + +# ── Python environment ─────────────────────────────────────────────── + +resource "fission_environment" "python" { + name = "weather-python" + image = "naeel/fission-python-env:v1.0" + version = 2 + namespace = var.namespace +} + +# ── Пакет: fetcher ─────────────────────────────────────────────────── + +resource "fission_package" "fetcher" { + name = "weather-fetcher-pkg" + environment = fission_environment.python.name + namespace = var.namespace + source_dir = "${path.module}/fetcher" + deploy_type = "source" +} + +# ── Пакет: consumer ────────────────────────────────────────────────── + +resource "fission_package" "consumer" { + name = "weather-consumer-pkg" + environment = fission_environment.python.name + namespace = var.namespace + source_dir = "${path.module}/consumer" + deploy_type = "source" +} + +# ── Функция: fetcher (читает OWM, пишет в SQS) ─────────────────────── + +resource "fission_function" "fetcher" { + name = "weather-fetcher" + environment = fission_environment.python.name + namespace = var.namespace + package_name = fission_package.fetcher.name + entrypoint = "main" + # Env vars с секретами задаются через K8s Secret вне Terraform. + # Имя секрета: weather-fetcher-env (namespace: var.namespace) + # Ключи: OWM_API_KEY, SQS_ACCESS_KEY, SQS_SECRET_KEY +} + +# ── CRON trigger: каждые 10 минут ──────────────────────────────────── + +resource "fission_cron_trigger" "fetcher" { + name = "weather-cron" + function = fission_function.fetcher.name + namespace = var.namespace + cron = "*/10 * * * *" +} + +# ── Функция: consumer (читает из SQS, пишет в PG) ──────────────────── + +resource "fission_function" "consumer" { + name = "weather-consumer" + environment = fission_environment.python.name + namespace = var.namespace + package_name = fission_package.consumer.name + entrypoint = "main" + # Env vars: PG_DSN задаётся через K8s Secret weather-consumer-env +} + +# ── MQ trigger: SQS queue → consumer function ──────────────────────── + +resource "fission_mq_trigger" "weather" { + name = "weather-mq" + function = fission_function.consumer.name + namespace = var.namespace + queue = "weather-data" + access_key = var.sqs_access_key + secret_key = var.sqs_secret_key + sqs_endpoint = "http://shared-sqs.shared-sqs.svc.cluster.local:4100" +} + +# ── Outputs ────────────────────────────────────────────────────────── + +output "iot_device_phase" { + value = fission_iot_device.weather_station.phase + description = "Статус IoT-устройства weather-station." +} + +output "iot_mqtt_username" { + value = fission_iot_device.weather_station.mqtt_username + description = "MQTT username для weather-station." +} + +output "iot_topic_prefix" { + value = fission_iot_device.weather_station.topic_prefix + description = "MQTT topic prefix для weather-station." +} diff --git a/examples/weather-demo/migration.sql b/examples/weather-demo/migration.sql new file mode 100644 index 0000000..b5a4e7b --- /dev/null +++ b/examples/weather-demo/migration.sql @@ -0,0 +1,18 @@ +-- Миграция: таблица метрик погоды для weather-demo +-- Применять: psql $PG_DSN -f migration.sql + +CREATE TABLE IF NOT EXISTS weather_metrics ( + id BIGSERIAL PRIMARY KEY, + city TEXT NOT NULL, + country TEXT NOT NULL, + temperature NUMERIC(5,2), + feels_like NUMERIC(5,2), + humidity INTEGER, + pressure INTEGER, + wind_speed NUMERIC(6,2), + description TEXT, + recorded_at TIMESTAMPTZ NOT NULL DEFAULT now() +); + +CREATE INDEX IF NOT EXISTS idx_weather_city_time + ON weather_metrics (city, recorded_at DESC); diff --git a/examples/weather-demo/terraform.tfvars.example b/examples/weather-demo/terraform.tfvars.example new file mode 100644 index 0000000..9b6dbb8 --- /dev/null +++ b/examples/weather-demo/terraform.tfvars.example @@ -0,0 +1,10 @@ +# Скопируй в terraform.tfvars и заполни значения. +# НЕ коммитить файл с реальными секретами! + +kubeconfig_path = "/home/naeel/.kube/config" +namespace = "fission-weather" + +owm_api_key = "YOUR_OPENWEATHERMAP_API_KEY" +sqs_access_key = "SSAK-a9964f2723bc6d347f48d153" +sqs_secret_key = "YOUR_SQS_SECRET_KEY" +pg_dsn = "postgresql://super:PASSWORD@postgresqlk8s-master.dc5db45d-f8b4-4fd0-ad33-ec4dd017f2d5.svc.cluster.local:5432/sqsdb?sslmode=disable" diff --git a/terraform/provider/internal/client/client.go b/terraform/provider/internal/client/client.go index cdd4e8b..6da00b7 100644 --- a/terraform/provider/internal/client/client.go +++ b/terraform/provider/internal/client/client.go @@ -4,7 +4,10 @@ import ( "context" "fmt" + appsv1 "k8s.io/api/apps/v1" + corev1 "k8s.io/api/core/v1" apierrors "k8s.io/apimachinery/pkg/api/errors" + k8sresource "k8s.io/apimachinery/pkg/api/resource" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured" "k8s.io/apimachinery/pkg/runtime/schema" @@ -38,6 +41,28 @@ var httpTriggerGVR = schema.GroupVersionResource{ Resource: "httptriggers", } +var timeTriggerGVR = schema.GroupVersionResource{ + Group: "fission.io", + Version: "v1", + Resource: "timetriggers", +} + +var iotDeviceGVR = schema.GroupVersionResource{ + Group: "iot.kube5s.ru", + Version: "v1alpha1", + Resource: "iotdevices", +} + +const ( + sqsConsumerImage = "naeel/sqs-consumer:v1.0" + sqsDefaultEndpoint = "http://shared-sqs.shared-sqs.svc.cluster.local:4100" + fissionRouterBase = "http://router.fission.svc.cluster.local" + mqManagedByLabel = "app.kubernetes.io/managed-by" + mqManagedByVal = "fission-terraform" + mqComponentLabel = "component" + mqComponentVal = "mq-trigger" +) + // Client хранит клиентов Kubernetes API для работы с CRD Fission. type Client struct { DynClient dynamic.Interface @@ -253,3 +278,242 @@ func (c *Client) DeleteHTTPTrigger(ctx context.Context, namespace, name string) return nil } + +// ── MQ Trigger (K8s Deployment + Secret) ──────────────────────────── + +func mqSecretName(name string) string { return "sqs-mq-" + name } +func mqDeployName(name string) string { return "mq-" + name } + +// MQTriggerInfo — данные MQ trigger, считанные из K8s. +type MQTriggerInfo struct { + Function string + Queue string + SQSEndpoint string + DeployUID string +} + +// CreateMQTrigger создаёт Secret + Deployment для SQS consumer. +func (c *Client) CreateMQTrigger(ctx context.Context, name, ns, functionName, queue, sqsEndpoint, accessKey, secretKey string) (*MQTriggerInfo, error) { + if sqsEndpoint == "" { + sqsEndpoint = sqsDefaultEndpoint + } + functionURL := fissionRouterBase + "/" + functionName + secretName := mqSecretName(name) + deployName := mqDeployName(name) + + labels := map[string]string{ + mqManagedByLabel: mqManagedByVal, + mqComponentLabel: mqComponentVal, + "mq-trigger-name": name, + } + + secret := &corev1.Secret{ + ObjectMeta: metav1.ObjectMeta{Name: secretName, Namespace: ns, Labels: labels}, + StringData: map[string]string{ + "SQS_ACCESS_KEY": accessKey, + "SQS_SECRET_KEY": secretKey, + "SQS_ENDPOINT": sqsEndpoint, + }, + } + if _, err := c.K8sClient.CoreV1().Secrets(ns).Create(ctx, secret, metav1.CreateOptions{}); err != nil { + return nil, fmt.Errorf("create mq secret %q: %w", secretName, err) + } + + replicas := int32(1) + deploy := &appsv1.Deployment{ + ObjectMeta: metav1.ObjectMeta{ + Name: deployName, + Namespace: ns, + Labels: labels, + Annotations: map[string]string{ + "fission-terraform/mq-trigger-name": name, + "fission-terraform/function": functionName, + "fission-terraform/queue": queue, + "fission-terraform/sqs-endpoint": sqsEndpoint, + }, + }, + Spec: appsv1.DeploymentSpec{ + Replicas: &replicas, + Selector: &metav1.LabelSelector{MatchLabels: map[string]string{"mq-trigger-name": name}}, + Template: corev1.PodTemplateSpec{ + ObjectMeta: metav1.ObjectMeta{Labels: map[string]string{ + "mq-trigger-name": name, + mqComponentLabel: mqComponentVal, + }}, + Spec: corev1.PodSpec{ + Containers: []corev1.Container{{ + Name: "sqs-consumer", + Image: sqsConsumerImage, + ImagePullPolicy: corev1.PullAlways, + Env: []corev1.EnvVar{ + {Name: "SQS_QUEUE_NAME", Value: queue}, + {Name: "SQS_REGION", Value: "us-east-1"}, + {Name: "FUNCTION_URL", Value: functionURL}, + {Name: "POLL_INTERVAL", Value: "5"}, + {Name: "MAX_MESSAGES", Value: "1"}, + {Name: "MAX_RETRIES", Value: "3"}, + }, + EnvFrom: []corev1.EnvFromSource{{ + SecretRef: &corev1.SecretEnvSource{ + LocalObjectReference: corev1.LocalObjectReference{Name: secretName}, + }, + }}, + Resources: corev1.ResourceRequirements{ + Limits: corev1.ResourceList{ + corev1.ResourceCPU: k8sresource.MustParse("50m"), + corev1.ResourceMemory: k8sresource.MustParse("32Mi"), + }, + Requests: corev1.ResourceList{ + corev1.ResourceCPU: k8sresource.MustParse("10m"), + corev1.ResourceMemory: k8sresource.MustParse("16Mi"), + }, + }, + }}, + }, + }, + }, + } + + created, err := c.K8sClient.AppsV1().Deployments(ns).Create(ctx, deploy, metav1.CreateOptions{}) + if err != nil { + _ = c.K8sClient.CoreV1().Secrets(ns).Delete(ctx, secretName, metav1.DeleteOptions{}) + return nil, fmt.Errorf("create mq deployment %q: %w", deployName, err) + } + + return &MQTriggerInfo{ + Function: functionName, + Queue: queue, + SQSEndpoint: sqsEndpoint, + DeployUID: string(created.UID), + }, nil +} + +// GetMQTrigger возвращает информацию о MQ trigger по аннотациям Deployment. +func (c *Client) GetMQTrigger(ctx context.Context, ns, name string) (*MQTriggerInfo, error) { + deploy, err := c.K8sClient.AppsV1().Deployments(ns).Get(ctx, mqDeployName(name), metav1.GetOptions{}) + if err != nil { + return nil, fmt.Errorf("get mq deployment %q: %w", name, err) + } + ann := deploy.Annotations + return &MQTriggerInfo{ + Function: ann["fission-terraform/function"], + Queue: ann["fission-terraform/queue"], + SQSEndpoint: ann["fission-terraform/sqs-endpoint"], + DeployUID: string(deploy.UID), + }, nil +} + +// DeleteMQTrigger удаляет Deployment + Secret MQ trigger. +func (c *Client) DeleteMQTrigger(ctx context.Context, ns, name string) error { + dErr := c.K8sClient.AppsV1().Deployments(ns).Delete(ctx, mqDeployName(name), metav1.DeleteOptions{}) + sErr := c.K8sClient.CoreV1().Secrets(ns).Delete(ctx, mqSecretName(name), metav1.DeleteOptions{}) + if dErr != nil && !apierrors.IsNotFound(dErr) { + return fmt.Errorf("delete mq deployment %q: %w", name, dErr) + } + if sErr != nil && !apierrors.IsNotFound(sErr) { + return fmt.Errorf("delete mq secret %q: %w", name, sErr) + } + return nil +} + +// ── TimeTrigger (Fission CRD) ──────────────────────────────────────── + +// CreateTimeTrigger создаёт объект TimeTrigger (CRON) в заданном namespace. +func (c *Client) CreateTimeTrigger(ctx context.Context, tt *unstructured.Unstructured) (*unstructured.Unstructured, error) { + created, err := c.DynClient.Resource(timeTriggerGVR).Namespace(tt.GetNamespace()).Create(ctx, tt, metav1.CreateOptions{}) + if err != nil { + return nil, fmt.Errorf("create time trigger %q: %w", tt.GetName(), err) + } + return created, nil +} + +// GetTimeTrigger возвращает объект TimeTrigger по имени. +func (c *Client) GetTimeTrigger(ctx context.Context, namespace, name string) (*unstructured.Unstructured, error) { + tt, err := c.DynClient.Resource(timeTriggerGVR).Namespace(namespace).Get(ctx, name, metav1.GetOptions{}) + if err != nil { + return nil, fmt.Errorf("get time trigger %q: %w", name, err) + } + return tt, nil +} + +// UpdateTimeTrigger обновляет объект TimeTrigger. +func (c *Client) UpdateTimeTrigger(ctx context.Context, tt *unstructured.Unstructured) (*unstructured.Unstructured, error) { + updated, err := c.DynClient.Resource(timeTriggerGVR).Namespace(tt.GetNamespace()).Update(ctx, tt, metav1.UpdateOptions{}) + if err != nil { + return nil, fmt.Errorf("update time trigger %q: %w", tt.GetName(), err) + } + return updated, nil +} + +// DeleteTimeTrigger удаляет объект TimeTrigger по имени. +func (c *Client) DeleteTimeTrigger(ctx context.Context, namespace, name string) error { + if err := c.DynClient.Resource(timeTriggerGVR).Namespace(namespace).Delete(ctx, name, metav1.DeleteOptions{}); err != nil { + return fmt.Errorf("delete time trigger %q: %w", name, err) + } + return nil +} + +// ── IoTDevice (CRD iot.kube5s.ru/v1alpha1) ────────────────────────── + +// IoTDeviceInfo — данные IoTDevice из статуса CRD. +type IoTDeviceInfo struct { + Phase string + MQTTUsername string + SecretName string + TopicPrefix string +} + +// CreateIoTDevice создаёт объект IoTDevice. +func (c *Client) CreateIoTDevice(ctx context.Context, name, namespace, deviceID string, metadata map[string]string) (*IoTDeviceInfo, error) { + obj := &unstructured.Unstructured{ + Object: map[string]any{ + "apiVersion": "iot.kube5s.ru/v1alpha1", + "kind": "IoTDevice", + "metadata": map[string]any{ + "name": name, + "namespace": namespace, + }, + "spec": map[string]any{ + "deviceId": deviceID, + "enabled": true, + "metadata": metadata, + }, + }, + } + created, err := c.DynClient.Resource(iotDeviceGVR).Namespace(namespace).Create(ctx, obj, metav1.CreateOptions{}) + if err != nil { + return nil, fmt.Errorf("create iot device %q: %w", name, err) + } + return extractIoTDeviceInfo(created), nil +} + +// GetIoTDevice возвращает IoTDevice по имени. +func (c *Client) GetIoTDevice(ctx context.Context, namespace, name string) (*IoTDeviceInfo, error) { + obj, err := c.DynClient.Resource(iotDeviceGVR).Namespace(namespace).Get(ctx, name, metav1.GetOptions{}) + if err != nil { + return nil, fmt.Errorf("get iot device %q: %w", name, err) + } + return extractIoTDeviceInfo(obj), nil +} + +// DeleteIoTDevice удаляет IoTDevice по имени. +func (c *Client) DeleteIoTDevice(ctx context.Context, namespace, name string) error { + if err := c.DynClient.Resource(iotDeviceGVR).Namespace(namespace).Delete(ctx, name, metav1.DeleteOptions{}); err != nil { + return fmt.Errorf("delete iot device %q: %w", name, err) + } + return nil +} + +func extractIoTDeviceInfo(obj *unstructured.Unstructured) *IoTDeviceInfo { + status, _, _ := unstructured.NestedMap(obj.Object, "status") + phase, _ := status["phase"].(string) + mqttUsername, _ := status["mqttUsername"].(string) + secretName, _ := status["secretName"].(string) + topicPrefix, _ := status["topicPrefix"].(string) + return &IoTDeviceInfo{ + Phase: phase, + MQTTUsername: mqttUsername, + SecretName: secretName, + TopicPrefix: topicPrefix, + } +} diff --git a/terraform/provider/internal/provider/provider.go b/terraform/provider/internal/provider/provider.go index 02797cc..138e006 100644 --- a/terraform/provider/internal/provider/provider.go +++ b/terraform/provider/internal/provider/provider.go @@ -99,6 +99,9 @@ func (p *FissionProvider) Resources(_ context.Context) []func() resource.Resourc resources.NewFunctionResource, resources.NewHTTPTriggerResource, resources.NewSimpleFunctionResource, + resources.NewMQTriggerResource, + resources.NewCronTriggerResource, + resources.NewIoTDeviceResource, } } diff --git a/terraform/provider/internal/resources/cron_trigger_resource.go b/terraform/provider/internal/resources/cron_trigger_resource.go new file mode 100644 index 0000000..e746933 --- /dev/null +++ b/terraform/provider/internal/resources/cron_trigger_resource.go @@ -0,0 +1,208 @@ +package resources + +import ( + "context" + "fmt" + + "github.com/hashicorp/terraform-plugin-framework/resource" + "github.com/hashicorp/terraform-plugin-framework/resource/schema" + "github.com/hashicorp/terraform-plugin-framework/resource/schema/planmodifier" + "github.com/hashicorp/terraform-plugin-framework/resource/schema/stringplanmodifier" + "github.com/hashicorp/terraform-plugin-framework/types" + "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured" + + "terraform-provider-fission/internal/client" +) + +var _ resource.Resource = &CronTriggerResource{} + +// CronTriggerResource управляет Fission TimeTrigger (CRON). +type CronTriggerResource struct { + client *client.Client +} + +type cronTriggerResourceModel struct { + ID types.String `tfsdk:"id"` + Name types.String `tfsdk:"name"` + Namespace types.String `tfsdk:"namespace"` + Function types.String `tfsdk:"function"` + Cron types.String `tfsdk:"cron"` + UID types.String `tfsdk:"uid"` +} + +func NewCronTriggerResource() resource.Resource { + return &CronTriggerResource{} +} + +func (r *CronTriggerResource) Metadata(_ context.Context, req resource.MetadataRequest, resp *resource.MetadataResponse) { + resp.TypeName = req.ProviderTypeName + "_cron_trigger" +} + +func (r *CronTriggerResource) Schema(_ context.Context, _ resource.SchemaRequest, resp *resource.SchemaResponse) { + resp.Schema = schema.Schema{ + Description: "Fission TimeTrigger — запускает функцию по CRON-расписанию.", + Attributes: map[string]schema.Attribute{ + "id": schema.StringAttribute{ + Computed: true, + Description: "Идентификатор ресурса (namespace/name).", + }, + "name": schema.StringAttribute{ + Required: true, + Description: "Имя TimeTrigger.", + PlanModifiers: []planmodifier.String{ + stringplanmodifier.RequiresReplace(), + }, + }, + "namespace": schema.StringAttribute{ + Optional: true, + Computed: true, + Description: "Kubernetes namespace.", + PlanModifiers: []planmodifier.String{ + stringplanmodifier.RequiresReplace(), + }, + }, + "function": schema.StringAttribute{ + Required: true, + Description: "Имя Fission Function для запуска.", + }, + "cron": schema.StringAttribute{ + Required: true, + Description: "CRON-расписание, например \"*/10 * * * *\".", + }, + "uid": schema.StringAttribute{ + Computed: true, + Description: "UID объекта в Kubernetes.", + }, + }, + } +} + +func (r *CronTriggerResource) Configure(_ context.Context, req resource.ConfigureRequest, resp *resource.ConfigureResponse) { + if req.ProviderData == nil { + return + } + c, ok := req.ProviderData.(*client.Client) + if !ok { + resp.Diagnostics.AddError("Неверный тип провайдера", fmt.Sprintf("ожидался *client.Client, получен %T", req.ProviderData)) + return + } + r.client = c +} + +func (r *CronTriggerResource) Create(ctx context.Context, req resource.CreateRequest, resp *resource.CreateResponse) { + var plan cronTriggerResourceModel + resp.Diagnostics.Append(req.Plan.Get(ctx, &plan)...) + if resp.Diagnostics.HasError() { + return + } + + ns := plan.Namespace.ValueString() + if ns == "" { + ns = r.client.Namespace + } + + obj := buildTimeTrigger(plan.Name.ValueString(), ns, plan.Function.ValueString(), plan.Cron.ValueString()) + created, err := r.client.CreateTimeTrigger(ctx, obj) + if err != nil { + resp.Diagnostics.AddError("Ошибка создания CRON trigger", err.Error()) + return + } + + plan.ID = types.StringValue(ns + "/" + plan.Name.ValueString()) + plan.Namespace = types.StringValue(ns) + plan.UID = types.StringValue(string(created.GetUID())) + resp.Diagnostics.Append(resp.State.Set(ctx, &plan)...) +} + +func (r *CronTriggerResource) Read(ctx context.Context, req resource.ReadRequest, resp *resource.ReadResponse) { + var state cronTriggerResourceModel + resp.Diagnostics.Append(req.State.Get(ctx, &state)...) + if resp.Diagnostics.HasError() { + return + } + + obj, err := r.client.GetTimeTrigger(ctx, state.Namespace.ValueString(), state.Name.ValueString()) + if err != nil { + if client.IsNotFound(err) { + resp.State.RemoveResource(ctx) + return + } + resp.Diagnostics.AddError("Ошибка чтения CRON trigger", err.Error()) + return + } + + spec, _, _ := unstructured.NestedMap(obj.Object, "spec") + if cron, ok := spec["cron"].(string); ok { + state.Cron = types.StringValue(cron) + } + if fnRef, ok := spec["functionref"].(map[string]any); ok { + if name, ok := fnRef["name"].(string); ok { + state.Function = types.StringValue(name) + } + } + state.UID = types.StringValue(string(obj.GetUID())) + resp.Diagnostics.Append(resp.State.Set(ctx, &state)...) +} + +func (r *CronTriggerResource) Update(ctx context.Context, req resource.UpdateRequest, resp *resource.UpdateResponse) { + var plan cronTriggerResourceModel + resp.Diagnostics.Append(req.Plan.Get(ctx, &plan)...) + if resp.Diagnostics.HasError() { + return + } + + existing, err := r.client.GetTimeTrigger(ctx, plan.Namespace.ValueString(), plan.Name.ValueString()) + if err != nil { + resp.Diagnostics.AddError("Ошибка получения CRON trigger для обновления", err.Error()) + return + } + + if err := unstructured.SetNestedField(existing.Object, plan.Cron.ValueString(), "spec", "cron"); err != nil { + resp.Diagnostics.AddError("Ошибка обновления cron", err.Error()) + return + } + if err := unstructured.SetNestedField(existing.Object, plan.Function.ValueString(), "spec", "functionref", "name"); err != nil { + resp.Diagnostics.AddError("Ошибка обновления function", err.Error()) + return + } + + updated, err := r.client.UpdateTimeTrigger(ctx, existing) + if err != nil { + resp.Diagnostics.AddError("Ошибка обновления CRON trigger", err.Error()) + return + } + + plan.UID = types.StringValue(string(updated.GetUID())) + resp.Diagnostics.Append(resp.State.Set(ctx, &plan)...) +} + +func (r *CronTriggerResource) Delete(ctx context.Context, req resource.DeleteRequest, resp *resource.DeleteResponse) { + var state cronTriggerResourceModel + resp.Diagnostics.Append(req.State.Get(ctx, &state)...) + if resp.Diagnostics.HasError() { + return + } + if err := r.client.DeleteTimeTrigger(ctx, state.Namespace.ValueString(), state.Name.ValueString()); err != nil { + resp.Diagnostics.AddError("Ошибка удаления CRON trigger", err.Error()) + } +} + +func buildTimeTrigger(name, ns, functionName, cron string) *unstructured.Unstructured { + return &unstructured.Unstructured{ + Object: map[string]any{ + "apiVersion": "fission.io/v1", + "kind": "TimeTrigger", + "metadata": map[string]any{ + "name": name, + "namespace": ns, + }, + "spec": map[string]any{ + "cron": cron, + "functionref": map[string]any{ + "type": "name", + "name": functionName, + }, + }, + }, + } +} diff --git a/terraform/provider/internal/resources/iot_device_resource.go b/terraform/provider/internal/resources/iot_device_resource.go new file mode 100644 index 0000000..140eaf4 --- /dev/null +++ b/terraform/provider/internal/resources/iot_device_resource.go @@ -0,0 +1,230 @@ +package resources + +import ( + "context" + "fmt" + + "github.com/hashicorp/terraform-plugin-framework/attr" + "github.com/hashicorp/terraform-plugin-framework/resource" + "github.com/hashicorp/terraform-plugin-framework/resource/schema" + "github.com/hashicorp/terraform-plugin-framework/resource/schema/booldefault" + "github.com/hashicorp/terraform-plugin-framework/resource/schema/mapdefault" + "github.com/hashicorp/terraform-plugin-framework/resource/schema/planmodifier" + "github.com/hashicorp/terraform-plugin-framework/resource/schema/stringdefault" + "github.com/hashicorp/terraform-plugin-framework/resource/schema/stringplanmodifier" + "github.com/hashicorp/terraform-plugin-framework/types" + + "terraform-provider-fission/internal/client" +) + +var _ resource.Resource = &IoTDeviceResource{} + +// IoTDeviceResource управляет CRD IoTDevice (iot.kube5s.ru/v1alpha1). +type IoTDeviceResource struct { + client *client.Client +} + +type iotDeviceResourceModel struct { + ID types.String `tfsdk:"id"` + Name types.String `tfsdk:"name"` + Namespace types.String `tfsdk:"namespace"` + DeviceID types.String `tfsdk:"device_id"` + Enabled types.Bool `tfsdk:"enabled"` + Metadata types.Map `tfsdk:"metadata"` + Phase types.String `tfsdk:"phase"` + MQTTUsername types.String `tfsdk:"mqtt_username"` + SecretName types.String `tfsdk:"secret_name"` + TopicPrefix types.String `tfsdk:"topic_prefix"` +} + +func NewIoTDeviceResource() resource.Resource { + return &IoTDeviceResource{} +} + +func (r *IoTDeviceResource) Metadata(_ context.Context, req resource.MetadataRequest, resp *resource.MetadataResponse) { + resp.TypeName = req.ProviderTypeName + "_iot_device" +} + +func (r *IoTDeviceResource) Schema(_ context.Context, _ resource.SchemaRequest, resp *resource.SchemaResponse) { + resp.Schema = schema.Schema{ + Description: "IoT-устройство, зарегистрированное в платформе sless (CRD iot.kube5s.ru/v1alpha1/IoTDevice).", + Attributes: map[string]schema.Attribute{ + "id": schema.StringAttribute{ + Computed: true, + Description: "Идентификатор ресурса (namespace/name).", + }, + "name": schema.StringAttribute{ + Required: true, + Description: "Имя IoTDevice.", + PlanModifiers: []planmodifier.String{ + stringplanmodifier.RequiresReplace(), + }, + }, + "namespace": schema.StringAttribute{ + Optional: true, + Computed: true, + Default: stringdefault.StaticString("sless"), + Description: "Namespace IoTDevice. По умолчанию sless.", + PlanModifiers: []planmodifier.String{ + stringplanmodifier.RequiresReplace(), + }, + }, + "device_id": schema.StringAttribute{ + Required: true, + Description: "Уникальный ID устройства внутри namespace (строчные буквы, цифры, дефис).", + PlanModifiers: []planmodifier.String{ + stringplanmodifier.RequiresReplace(), + }, + }, + "enabled": schema.BoolAttribute{ + Optional: true, + Computed: true, + Default: booldefault.StaticBool(true), + Description: "Активно ли устройство.", + }, + "metadata": schema.MapAttribute{ + Optional: true, + Computed: true, + ElementType: types.StringType, + Default: mapdefault.StaticValue(types.MapValueMust(types.StringType, map[string]attr.Value{})), + Description: "Произвольные метаданные устройства (модель, локация и т.д.).", + }, + // Computed (заполняет iot-operator) + "phase": schema.StringAttribute{ + Computed: true, + Description: "Текущая фаза: Active, Disabled, Pending, Error.", + }, + "mqtt_username": schema.StringAttribute{ + Computed: true, + Description: "MQTT username, выданный iot-operator.", + }, + "secret_name": schema.StringAttribute{ + Computed: true, + Description: "Имя K8s Secret с MQTT credentials.", + }, + "topic_prefix": schema.StringAttribute{ + Computed: true, + Description: "MQTT topic prefix для публикации.", + }, + }, + } +} + +func (r *IoTDeviceResource) Configure(_ context.Context, req resource.ConfigureRequest, resp *resource.ConfigureResponse) { + if req.ProviderData == nil { + return + } + c, ok := req.ProviderData.(*client.Client) + if !ok { + resp.Diagnostics.AddError("Неверный тип провайдера", fmt.Sprintf("ожидался *client.Client, получен %T", req.ProviderData)) + return + } + r.client = c +} + +func (r *IoTDeviceResource) Create(ctx context.Context, req resource.CreateRequest, resp *resource.CreateResponse) { + var plan iotDeviceResourceModel + resp.Diagnostics.Append(req.Plan.Get(ctx, &plan)...) + if resp.Diagnostics.HasError() { + return + } + + ns := plan.Namespace.ValueString() + meta := map[string]string{} + if !plan.Metadata.IsNull() && !plan.Metadata.IsUnknown() { + elems := plan.Metadata.Elements() + for k, v := range elems { + if sv, ok := v.(types.String); ok { + meta[k] = sv.ValueString() + } + } + } + + info, err := r.client.CreateIoTDevice(ctx, plan.Name.ValueString(), ns, plan.DeviceID.ValueString(), meta) + if err != nil { + resp.Diagnostics.AddError("Ошибка создания IoT device", err.Error()) + return + } + + plan.ID = types.StringValue(ns + "/" + plan.Name.ValueString()) + plan.Phase = types.StringValue(info.Phase) + plan.MQTTUsername = types.StringValue(info.MQTTUsername) + plan.SecretName = types.StringValue(info.SecretName) + plan.TopicPrefix = types.StringValue(info.TopicPrefix) + resp.Diagnostics.Append(resp.State.Set(ctx, &plan)...) +} + +func (r *IoTDeviceResource) Read(ctx context.Context, req resource.ReadRequest, resp *resource.ReadResponse) { + var state iotDeviceResourceModel + resp.Diagnostics.Append(req.State.Get(ctx, &state)...) + if resp.Diagnostics.HasError() { + return + } + + info, err := r.client.GetIoTDevice(ctx, state.Namespace.ValueString(), state.Name.ValueString()) + if err != nil { + if client.IsNotFound(err) { + resp.State.RemoveResource(ctx) + return + } + resp.Diagnostics.AddError("Ошибка чтения IoT device", err.Error()) + return + } + + state.Phase = types.StringValue(info.Phase) + state.MQTTUsername = types.StringValue(info.MQTTUsername) + state.SecretName = types.StringValue(info.SecretName) + state.TopicPrefix = types.StringValue(info.TopicPrefix) + resp.Diagnostics.Append(resp.State.Set(ctx, &state)...) +} + +func (r *IoTDeviceResource) Update(ctx context.Context, req resource.UpdateRequest, resp *resource.UpdateResponse) { + // name/namespace/device_id — RequiresReplace, значит Update только для enabled/metadata. + // Для простоты: пересоздаём объект. + var plan iotDeviceResourceModel + resp.Diagnostics.Append(req.Plan.Get(ctx, &plan)...) + if resp.Diagnostics.HasError() { + return + } + + ns := plan.Namespace.ValueString() + name := plan.Name.ValueString() + + if err := r.client.DeleteIoTDevice(ctx, ns, name); err != nil && !client.IsNotFound(err) { + resp.Diagnostics.AddError("Ошибка удаления IoT device при обновлении", err.Error()) + return + } + + meta := map[string]string{} + if !plan.Metadata.IsNull() && !plan.Metadata.IsUnknown() { + elems := plan.Metadata.Elements() + for k, v := range elems { + if sv, ok := v.(types.String); ok { + meta[k] = sv.ValueString() + } + } + } + + info, err := r.client.CreateIoTDevice(ctx, name, ns, plan.DeviceID.ValueString(), meta) + if err != nil { + resp.Diagnostics.AddError("Ошибка создания IoT device при обновлении", err.Error()) + return + } + + plan.Phase = types.StringValue(info.Phase) + plan.MQTTUsername = types.StringValue(info.MQTTUsername) + plan.SecretName = types.StringValue(info.SecretName) + plan.TopicPrefix = types.StringValue(info.TopicPrefix) + resp.Diagnostics.Append(resp.State.Set(ctx, &plan)...) +} + +func (r *IoTDeviceResource) Delete(ctx context.Context, req resource.DeleteRequest, resp *resource.DeleteResponse) { + var state iotDeviceResourceModel + resp.Diagnostics.Append(req.State.Get(ctx, &state)...) + if resp.Diagnostics.HasError() { + return + } + if err := r.client.DeleteIoTDevice(ctx, state.Namespace.ValueString(), state.Name.ValueString()); err != nil { + resp.Diagnostics.AddError("Ошибка удаления IoT device", err.Error()) + } +} diff --git a/terraform/provider/internal/resources/mq_trigger_resource.go b/terraform/provider/internal/resources/mq_trigger_resource.go new file mode 100644 index 0000000..0facdb4 --- /dev/null +++ b/terraform/provider/internal/resources/mq_trigger_resource.go @@ -0,0 +1,205 @@ +package resources + +import ( + "context" + "fmt" + + "github.com/hashicorp/terraform-plugin-framework/resource" + "github.com/hashicorp/terraform-plugin-framework/resource/schema" + "github.com/hashicorp/terraform-plugin-framework/resource/schema/planmodifier" + "github.com/hashicorp/terraform-plugin-framework/resource/schema/stringdefault" + "github.com/hashicorp/terraform-plugin-framework/resource/schema/stringplanmodifier" + "github.com/hashicorp/terraform-plugin-framework/types" + + "terraform-provider-fission/internal/client" +) + +var _ resource.Resource = &MQTriggerResource{} + +// MQTriggerResource управляет MQ-триггером через K8s Deployment+Secret (sqs-consumer). +type MQTriggerResource struct { + client *client.Client +} + +type mqTriggerResourceModel struct { + ID types.String `tfsdk:"id"` + Name types.String `tfsdk:"name"` + Namespace types.String `tfsdk:"namespace"` + Function types.String `tfsdk:"function"` + Queue types.String `tfsdk:"queue"` + AccessKey types.String `tfsdk:"access_key"` + SecretKey types.String `tfsdk:"secret_key"` + SQSEndpoint types.String `tfsdk:"sqs_endpoint"` +} + +func NewMQTriggerResource() resource.Resource { + return &MQTriggerResource{} +} + +func (r *MQTriggerResource) Metadata(_ context.Context, req resource.MetadataRequest, resp *resource.MetadataResponse) { + resp.TypeName = req.ProviderTypeName + "_mq_trigger" +} + +func (r *MQTriggerResource) Schema(_ context.Context, _ resource.SchemaRequest, resp *resource.SchemaResponse) { + resp.Schema = schema.Schema{ + Description: "MQ-триггер через K8s Deployment (sqs-consumer) + Secret с SQS credentials.", + Attributes: map[string]schema.Attribute{ + "id": schema.StringAttribute{ + Computed: true, + Description: "Идентификатор ресурса (namespace/name).", + }, + "name": schema.StringAttribute{ + Required: true, + Description: "Имя MQ trigger.", + PlanModifiers: []planmodifier.String{ + stringplanmodifier.RequiresReplace(), + }, + }, + "namespace": schema.StringAttribute{ + Optional: true, + Computed: true, + Description: "Kubernetes namespace.", + PlanModifiers: []planmodifier.String{ + stringplanmodifier.RequiresReplace(), + }, + }, + "function": schema.StringAttribute{ + Required: true, + Description: "Имя Fission Function, к которой направляются сообщения из очереди.", + }, + "queue": schema.StringAttribute{ + Required: true, + Description: "Имя SQS очереди.", + }, + "access_key": schema.StringAttribute{ + Required: true, + Sensitive: true, + Description: "SQS Access Key.", + }, + "secret_key": schema.StringAttribute{ + Required: true, + Sensitive: true, + Description: "SQS Secret Key.", + }, + "sqs_endpoint": schema.StringAttribute{ + Optional: true, + Computed: true, + Default: stringdefault.StaticString("http://shared-sqs.shared-sqs.svc.cluster.local:4100"), + Description: "SQS endpoint URL. По умолчанию — внутренний shared-sqs.", + }, + }, + } +} + +func (r *MQTriggerResource) Configure(_ context.Context, req resource.ConfigureRequest, resp *resource.ConfigureResponse) { + if req.ProviderData == nil { + return + } + c, ok := req.ProviderData.(*client.Client) + if !ok { + resp.Diagnostics.AddError("Неверный тип провайдера", fmt.Sprintf("ожидался *client.Client, получен %T", req.ProviderData)) + return + } + r.client = c +} + +func (r *MQTriggerResource) Create(ctx context.Context, req resource.CreateRequest, resp *resource.CreateResponse) { + var plan mqTriggerResourceModel + resp.Diagnostics.Append(req.Plan.Get(ctx, &plan)...) + if resp.Diagnostics.HasError() { + return + } + + ns := plan.Namespace.ValueString() + if ns == "" { + ns = r.client.Namespace + } + + info, err := r.client.CreateMQTrigger( + ctx, + plan.Name.ValueString(), + ns, + plan.Function.ValueString(), + plan.Queue.ValueString(), + plan.SQSEndpoint.ValueString(), + plan.AccessKey.ValueString(), + plan.SecretKey.ValueString(), + ) + if err != nil { + resp.Diagnostics.AddError("Ошибка создания MQ trigger", err.Error()) + return + } + + plan.ID = types.StringValue(ns + "/" + plan.Name.ValueString()) + plan.Namespace = types.StringValue(ns) + plan.SQSEndpoint = types.StringValue(info.SQSEndpoint) + resp.Diagnostics.Append(resp.State.Set(ctx, &plan)...) +} + +func (r *MQTriggerResource) Read(ctx context.Context, req resource.ReadRequest, resp *resource.ReadResponse) { + var state mqTriggerResourceModel + resp.Diagnostics.Append(req.State.Get(ctx, &state)...) + if resp.Diagnostics.HasError() { + return + } + + ns := state.Namespace.ValueString() + info, err := r.client.GetMQTrigger(ctx, ns, state.Name.ValueString()) + if err != nil { + if client.IsNotFound(err) { + resp.State.RemoveResource(ctx) + return + } + resp.Diagnostics.AddError("Ошибка чтения MQ trigger", err.Error()) + return + } + + state.Function = types.StringValue(info.Function) + state.Queue = types.StringValue(info.Queue) + state.SQSEndpoint = types.StringValue(info.SQSEndpoint) + resp.Diagnostics.Append(resp.State.Set(ctx, &state)...) +} + +func (r *MQTriggerResource) Update(ctx context.Context, req resource.UpdateRequest, resp *resource.UpdateResponse) { + var plan mqTriggerResourceModel + resp.Diagnostics.Append(req.Plan.Get(ctx, &plan)...) + if resp.Diagnostics.HasError() { + return + } + + ns := plan.Namespace.ValueString() + name := plan.Name.ValueString() + + // Удаляем старый и создаём заново с новыми параметрами. + if err := r.client.DeleteMQTrigger(ctx, ns, name); err != nil { + resp.Diagnostics.AddError("Ошибка удаления MQ trigger при обновлении", err.Error()) + return + } + info, err := r.client.CreateMQTrigger( + ctx, name, ns, + plan.Function.ValueString(), + plan.Queue.ValueString(), + plan.SQSEndpoint.ValueString(), + plan.AccessKey.ValueString(), + plan.SecretKey.ValueString(), + ) + if err != nil { + resp.Diagnostics.AddError("Ошибка создания MQ trigger при обновлении", err.Error()) + return + } + + plan.SQSEndpoint = types.StringValue(info.SQSEndpoint) + resp.Diagnostics.Append(resp.State.Set(ctx, &plan)...) +} + +func (r *MQTriggerResource) Delete(ctx context.Context, req resource.DeleteRequest, resp *resource.DeleteResponse) { + var state mqTriggerResourceModel + resp.Diagnostics.Append(req.State.Get(ctx, &state)...) + if resp.Diagnostics.HasError() { + return + } + + if err := r.client.DeleteMQTrigger(ctx, state.Namespace.ValueString(), state.Name.ValueString()); err != nil { + resp.Diagnostics.AddError("Ошибка удаления MQ trigger", err.Error()) + } +}