feat(provider): add fission_mq_trigger, fission_cron_trigger, fission_iot_device + weather-demo example
This commit is contained in:
“Naeel”
@@ -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,
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user