feat: REST API (gorilla/mux + slog) + trigger controller + main.go wiring

- controllers/trigger_controller.go: полная реализация, HTTP->Ingress+Service, Cron->CronJob
- internal/config: добавлены IngressHost, APIToken
- internal/api/router.go: gorilla/mux роутер /v1/namespaces/{ns}/...
- internal/api/handler/: functions, triggers, invocations, base handler с slog
- internal/api/middleware/: auth (Bearer token) + logging (slog)
- main.go: запуск operator + HTTP сервера параллельно
- go.sum: добавлен gorilla/mux v1.8.1
This commit is contained in:
“Naeel”
2026-03-07 08:46:10 +04:00
parent 5f25f61948
commit 43e7d0ea48
21 changed files with 1720 additions and 112 deletions
View File
+108
View File
@@ -0,0 +1,108 @@
// Изменено: 2026-03-07
// PostgreSQL storage — хранение логов вызовов функций.
// Состояние функций (phase, imageRef) хранится в k8s CRD, не здесь.
// Здесь только invocations: логи, статусы, время выполнения.
package postgres
import (
"context"
"database/sql"
"fmt"
"time"
_ "github.com/lib/pq" // драйвер PostgreSQL
)
// Invocation — запись о вызове функции.
type Invocation struct {
ID string
FunctionName string
Namespace string
Status string // success, error, timeout
DurationMs int32
HTTPStatus *int32 // nil для cron триггеров
Logs string
TriggerType string
CreatedAt time.Time
}
// Store — клиент для работы с PostgreSQL.
type Store struct {
db *sql.DB
}
// New открывает подключение к PostgreSQL и проверяет его.
func New(dsn string) (*Store, error) {
db, err := sql.Open("postgres", dsn)
if err != nil {
return nil, fmt.Errorf("open postgres: %w", err)
}
// Проверяем что подключение реально работает
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
if err := db.PingContext(ctx); err != nil {
return nil, fmt.Errorf("ping postgres: %w", err)
}
db.SetMaxOpenConns(10)
db.SetMaxIdleConns(5)
db.SetConnMaxLifetime(5 * time.Minute)
return &Store{db: db}, nil
}
// Close закрывает подключение к БД.
func (s *Store) Close() error {
return s.db.Close()
}
// SaveInvocation записывает лог вызова функции в БД.
func (s *Store) SaveInvocation(ctx context.Context, inv *Invocation) error {
_, err := s.db.ExecContext(ctx, `
INSERT INTO invocations (function_name, namespace, status, duration_ms, http_status, logs, trigger_type)
VALUES ($1, $2, $3, $4, $5, $6, $7)
`, inv.FunctionName, inv.Namespace, inv.Status, inv.DurationMs, inv.HTTPStatus, inv.Logs, inv.TriggerType)
if err != nil {
return fmt.Errorf("save invocation: %w", err)
}
return nil
}
// ListInvocations возвращает последние N вызовов для указанной функции.
func (s *Store) ListInvocations(ctx context.Context, functionName, namespace string, limit int) ([]*Invocation, error) {
rows, err := s.db.QueryContext(ctx, `
SELECT id, function_name, namespace, status, duration_ms, http_status, logs, trigger_type, created_at
FROM invocations
WHERE function_name = $1 AND namespace = $2
ORDER BY created_at DESC
LIMIT $3
`, functionName, namespace, limit)
if err != nil {
return nil, fmt.Errorf("list invocations: %w", err)
}
defer rows.Close()
var result []*Invocation
for rows.Next() {
inv := &Invocation{}
if err := rows.Scan(
&inv.ID, &inv.FunctionName, &inv.Namespace,
&inv.Status, &inv.DurationMs, &inv.HTTPStatus,
&inv.Logs, &inv.TriggerType, &inv.CreatedAt,
); err != nil {
return nil, fmt.Errorf("scan invocation: %w", err)
}
result = append(result, inv)
}
return result, rows.Err()
}
// RunMigrations применяет SQL файлы из директории migrations.
// Простая реализация без внешних зависимостей — выполняем один файл.
// Используем IF NOT EXISTS в SQL поэтому безопасно запускать повторно.
func (s *Store) RunMigrations(ctx context.Context, sql string) error {
_, err := s.db.ExecContext(ctx, sql)
if err != nil {
return fmt.Errorf("run migration: %w", err)
}
return nil
}
View File
+82
View File
@@ -0,0 +1,82 @@
// Изменено: 2026-03-07
// S3 storage — загрузка и скачивание zip архивов с кодом функций.
// Используем Ceph S3 compatible API (minio-go клиент умеет работать с любым S3).
// Код загружается пользователем через REST API, хранится в S3,
// builder скачивает его для сборки Docker образа.
package s3
import (
"context"
"fmt"
"io"
"github.com/minio/minio-go/v7"
"github.com/minio/minio-go/v7/pkg/credentials"
)
// Client — клиент для работы с S3.
type Client struct {
mc *minio.Client
bucket string
}
// New создаёт клиент S3 и проверяет/создаёт бакет.
func New(endpoint, accessKey, secretKey, bucket string, useSSL bool) (*Client, error) {
mc, err := minio.New(endpoint, &minio.Options{
Creds: credentials.NewStaticV4(accessKey, secretKey, ""),
Secure: useSSL,
})
if err != nil {
return nil, fmt.Errorf("create minio client: %w", err)
}
return &Client{mc: mc, bucket: bucket}, nil
}
// EnsureBucket создаёт бакет если он не существует.
// Вызывается при старте сервиса.
func (c *Client) EnsureBucket(ctx context.Context) error {
exists, err := c.mc.BucketExists(ctx, c.bucket)
if err != nil {
return fmt.Errorf("check bucket: %w", err)
}
if !exists {
if err := c.mc.MakeBucket(ctx, c.bucket, minio.MakeBucketOptions{}); err != nil {
return fmt.Errorf("create bucket: %w", err)
}
}
return nil
}
// Upload загружает zip архив с кодом функции в S3.
// Ключ: functions/{namespace}/{name}/{version}.zip
// Возвращает ключ объекта для сохранения в CRD.
func (c *Client) Upload(ctx context.Context, namespace, funcName, version string, r io.Reader, size int64) (string, error) {
key := fmt.Sprintf("functions/%s/%s/%s.zip", namespace, funcName, version)
_, err := c.mc.PutObject(ctx, c.bucket, key, r, size, minio.PutObjectOptions{
ContentType: "application/zip",
})
if err != nil {
return "", fmt.Errorf("upload function code: %w", err)
}
return key, nil
}
// Download скачивает zip архив с кодом функции из S3.
// Используется builder'ом для сборки образа.
func (c *Client) Download(ctx context.Context, key string) (io.ReadCloser, error) {
obj, err := c.mc.GetObject(ctx, c.bucket, key, minio.GetObjectOptions{})
if err != nil {
return nil, fmt.Errorf("download function code: %w", err)
}
return obj, nil
}
// Delete удаляет архив с кодом функции из S3.
// Вызывается при удалении Function CRD.
func (c *Client) Delete(ctx context.Context, key string) error {
if err := c.mc.RemoveObject(ctx, c.bucket, key, minio.RemoveObjectOptions{}); err != nil {
return fmt.Errorf("delete function code: %w", err)
}
return nil
}