server:
  port: 8080
  log_level: info
database:
  dsn: postgres://user:pass@localhost:5432/deploylite
redis:
  addr: redis:6379
  db: 0
storage:
  provider: s3
  endpoint: http://minio:9000
  bucket: artifacts
security:
  jwt_secret: "CHANGE_ME"

POST /api/v1/pipelines
{
  "project_id": 12,
  "yaml": "stages: [...]"
}

{
  "id": 88,
  "status": "pending",
  "created_at": "2025-10-22T10:00:00Z"
}

POST /api/v1/pipelines/88/run

GET /api/v1/pipelines/88/logs

type Scheduler struct {
    redis   *redis.Client
    runners map[string]*RunnerInfo
}

func (s *Scheduler) DispatchTask(task *Task) error {
    // 基于标签和负载选择 Runner
    runner := s.SelectRunner(task)
    s.redis.Publish("runner:"+runner.ID, task)
    return nil
}

input := map[string]any{
    "user": user,
    "action": "deploy",
    "project": projectID,
}
allowed, err := opaEngine.Evaluate("deploy.rego", input)
if !allowed {
    return errors.New("policy denied")
}

package deploylite.policy

default allow = false

allow {
  input.user.role == "maintainer"
  input.action == "deploy"
  input.env != "prod"
}

func (r *Runner) Register() error {
    req := map[string]string{"name": r.Name, "token": r.Token}
    resp, _ := http.Post(apiURL+"/register", req)
    return resp.StatusCode == 200
}

runner:
  mode: docker
  concurrency: 4
  labels: ["build", "linux"]

func (e *Executor) Run(cmd string) error {
    ctx, cancel := context.WithTimeout(context.Background(), e.Timeout)
    defer cancel()
    c := exec.CommandContext(ctx, "/bin/sh", "-c", cmd)
    c.Stdout = e.LogWriter
    c.Stderr = e.LogWriter
    return c.Run()
}

cachePath := filepath.Join(r.WorkDir, ".cache")
if _, err := os.Stat(cachePath); os.IsNotExist(err) {
    os.MkdirAll(cachePath, 0755)
}

version: 1
name: build-and-deploy
env:
  GO_VERSION: 1.22
stages:
  - name: build
    steps:
      - run: go build -o bin/app ./cmd
      - artifact:
          upload: bin/app
  - name: deploy
    needs: [build]
    steps:
      - run: scp bin/app user@server:/opt/app
      - run: ssh user@server 'systemctl restart app'

type Pipeline struct {
    Version int
    Name    string
    Env     map[string]string
    Stages  []Stage
}

type Stage struct {
    Name   string
    Needs  []string
    Steps  []Step
}

func ParseYAML(data []byte) (*Pipeline, error) {
    var p Pipeline
    if err := yaml.Unmarshal(data, &p); err != nil {
        return nil, err
    }
    return &p, validate(p)
}

type Node struct {
    Name     string
    Children []*Node
    Parents  []*Node
}

type DAG struct {
    Nodes map[string]*Node
}

func (g *DAG) AddEdge(a, b string) {
    g.Nodes[a].Children = append(g.Nodes[a].Children, g.Nodes[b])
    g.Nodes[b].Parents = append(g.Nodes[b].Parents, g.Nodes[a])
}

func (g *DAG) TopoSort() ([]*Node, error) {
    inDegree := make(map[string]int)
    for _, n := range g.Nodes {
        for _, c := range n.Children {
            inDegree[c.Name]++
        }
    }

    queue := []*Node{}
    for _, n := range g.Nodes {
        if inDegree[n.Name] == 0 {
            queue = append(queue, n)
        }
    }

    var result []*Node
    for len(queue) > 0 {
        node := queue[0]
        queue = queue[1:]
        result = append(result, node)
        for _, c := range node.Children {
            inDegree[c.Name]--
            if inDegree[c.Name] == 0 {
                queue = append(queue, c)
            }
        }
    }

    if len(result) != len(g.Nodes) {
        return nil, errors.New("circular dependency detected")
    }
    return result, nil
}

func (e *Engine) Run(ctx context.Context, p *Pipeline) error {
    dag, _ := BuildDAG(p)
    for _, stage := range dag.EntryPoints() {
        go e.runStage(ctx, stage)
    }
    <-e.done
    return nil
}

func (e *Engine) runStage(ctx context.Context, s *Stage) {
    for _, step := range s.Steps {
        e.runStep(ctx, step)
    }
    e.markStageDone(s.Name)
}

type Status string
const (
    StatusPending  Status = "pending"
    StatusRunning  Status = "running"
    StatusSuccess  Status = "success"
    StatusFailed   Status = "failed"
    StatusSkipped  Status = "skipped"
)

stages:
  - name: deploy
    retry: 3
    rollback:
      - run: ssh user@server 'systemctl stop app && mv old/app app'

func (e *Engine) rollback(stage Stage) {
    for _, step := range stage.Rollback {
        e.runStep(context.Background(), step)
    }
}

type Plugin interface {
    Name() string
    Execute(ctx Context, input map[string]any) error
}

registry.Register("docker.build", DockerBuildPlugin{})
registry.Register("k8s.deploy", K8sDeployPlugin{})

plugin := registry.Get(step.Type)
plugin.Execute(ctx, step.Params)

func (r *Reporter) Emit(event Event) {
    r.ws.Broadcast(event)
    r.kafka.Publish(event)
}

type Artifact struct {
    ID         int64
    ProjectID  int64
    Name       string
    Version    string
    Hash       string
    Size       int64
    StorageKey string
    CreatedAt  time.Time
}

func (s *Service) Upload(ctx context.Context, file io.Reader, meta Metadata) (Artifact, error) {
    hash := sha256.New()
    tee := io.TeeReader(file, hash)
    size, err := s.storage.Save(ctx, tee, meta.Path)
    if err != nil {
        return Artifact{}, err
    }
    art := Artifact{
        ProjectID:  meta.ProjectID,
        Name:       meta.Name,
        Version:    meta.Version,
        Size:       size,
        Hash:       hex.EncodeToString(hash.Sum(nil)),
        StorageKey: meta.Path,
    }
    return s.repo.Create(ctx, art)
}

url, _ := s.storage.Presign("artifacts/app-v1.2.tar.gz", 10*time.Minute)
return url

syft dir:./app -o json > sbom.json

cmd := exec.Command("syft", "dir:.", "-o", "json")
output, _ := cmd.Output()
s.repo.SaveSBOM(id, output)

func (s *Cleaner) Run() {
    arts := s.repo.FindOldArtifacts(30)
    for _, a := range arts {
        s.storage.Delete(a.StorageKey)
        s.repo.Delete(a.ID)
    }
}

cosign sign --key cosign.key artifacts/app-v1.2.tar.gz
cosign verify artifacts/app-v1.2.tar.gz

type ProjectRepo interface {
    Create(ctx context.Context, p *Project) error
    FindByID(ctx context.Context, id int64) (*Project, error)
    List(ctx context.Context) ([]Project, error)
}

type projectRepo struct{ db *gorm.DB }

func (r *projectRepo) Create(ctx context.Context, p *Project) error {
    return r.db.WithContext(ctx).Create(p).Error
}

func (c *Cache) GetOrSet(key string, fn func() (any, error)) (any, error) {
    if v, ok := c.local.Get(key); ok {
        return v, nil
    }
    v, err := fn()
    if err == nil {
        c.local.Set(key, v, 5*time.Minute)
        c.redis.Set(ctx, key, v, 10*time.Minute)
    }
    return v, err
}

tx := db.Begin()
if err := tx.Create(&pipeline).Error; err != nil {
    tx.Rollback()
}
if err := s.artifact.Save(tx, data); err != nil {
    tx.Rollback()
}
tx.Commit()

type Storage interface {
    Save(ctx context.Context, r io.Reader, key string) (int64, error)
    Delete(key string) error
    Presign(key string, ttl time.Duration) (string, error)
}

func (m *MinioStore) Save(ctx context.Context, r io.Reader, key string) (int64, error) {
    _, err := m.client.PutObject(ctx, m.bucket, key, r, -1, minio.PutObjectOptions{})
    return 0, err
}

CREATE INDEX idx_pipeline_status ON pipelines(status);
CREATE INDEX idx_artifact_project ON artifacts(project_id);
CREATE INDEX idx_runner_status ON runners(status);

var (
    PipelineDuration = prometheus.NewHistogramVec(
        prometheus.HistogramOpts{
            Name:    "deploylite_pipeline_duration_seconds",
            Help:    "Pipeline duration in seconds",
            Buckets: prometheus.LinearBuckets(10, 10, 10),
        },
        []string{"project", "status"},
    )

    RunnerActive = prometheus.NewGaugeVec(
        prometheus.GaugeOpts{
            Name: "deploylite_runner_active_total",
            Help: "Number of active runners",
        },
        []string{"region"},
    )
)

func InitMetrics() {
    prometheus.MustRegister(PipelineDuration, RunnerActive)
}

func CollectRunnerMetrics(runners []*Runner) {
    for _, r := range runners {
        RunnerActive.WithLabelValues(r.Region).Set(float64(r.ActiveTasks))
    }
}

{
  "time": "2025-10-23T11:12:00Z",
  "level": "info",
  "trace_id": "b33a-9f12-cc12",
  "module": "runner",
  "pipeline": "build-221",
  "message": "Build stage completed",
  "duration_ms": 4320
}

tp := sdktrace.NewTracerProvider(
    sdktrace.WithBatcher(exporter),
    sdktrace.WithResource(resource.NewWithAttributes(
        semconv.ServiceNameKey.String("deploylite-api"),
    )),
)
otel.SetTracerProvider(tp)

tracer := otel.Tracer("deploylite")
ctx, span := tracer.Start(ctx, "pipeline.run")
defer span.End()

groups:
  - name: deploylite.rules
    rules:
      - alert: RunnerOffline
        expr: deploylite_runner_active_total == 0
        for: 2m
        labels:
          severity: critical
        annotations:
          summary: "All runners are offline"

func SendAlert(event AlertEvent) {
    body, _ := json.Marshal(event)
    http.Post(AlertWebhook, "application/json", bytes.NewBuffer(body))
}

type Plugin interface {
    Name() string
    Version() string
    Execute(ctx Context, input map[string]any) (map[string]any, error)
}

registry.Register("notify.slack", SlackPlugin{})

cmd := exec.Command("python3", "plugin.py")
stdin, _ := cmd.StdinPipe()
stdout, _ := cmd.StdoutPipe()
go cmd.Start()
json.NewEncoder(stdin).Encode(input)
json.NewDecoder(stdout).Decode(&output)

import sys, json
data = json.load(sys.stdin)
print(json.dumps({"status": "ok"}))

type Registry struct {
    plugins map[string]Plugin
}

func (r *Registry) Register(name string, p Plugin) {
    r.plugins[name] = p
}

func (r *Registry) Get(name string) Plugin {
    return r.plugins[name]
}

name: docker.build
version: 1.0.1
author: "spcent"
type: step
inputs:
  - name: context
  - name: dockerfile

ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
cmd := exec.CommandContext(ctx, "plugin-exec", args...)

package plugin

func Run(p Plugin) {
    input := readInput()
    output, err := p.Execute(context.Background(), input)
    if err != nil {
        writeError(err)
    } else {
        writeOutput(output)
    }
}

type HelloPlugin struct{}
func (p HelloPlugin) Execute(ctx context.Context, in map[string]any) (map[string]any, error) {
    fmt.Println("Hello from plugin!")
    return map[string]any{"msg": "ok"}, nil
}

func GenerateToken(u *User) (string, error) {
    claims := jwt.MapClaims{
        "sub": u.ID,
        "role": u.Role,
        "exp": time.Now().Add(15 * time.Minute).Unix(),
    }
    token := jwt.NewWithClaims(jwt.SigningMethodHS256, claims)
    return token.SignedString([]byte(secret))
}

type Role struct {
    Name        string
    Permissions []string
}

roles:
  maintainer:
    - pipelines:create
    - pipelines:deploy
  viewer:
    - pipelines:read

package deploylite.policy

default allow = false

allow {
    input.user.role == "maintainer"
    input.action == "deploy"
    input.env != "prod"
}

decision, err := opa.Eval(ctx, rego.Query("data.deploylite.policy.allow"), rego.Input(input))

func Encrypt(data []byte, key []byte) ([]byte, error) {
    block, _ := aes.NewCipher(key)
    aesgcm, _ := cipher.NewGCM(block)
    nonce := make([]byte, aesgcm.NonceSize())
    rand.Read(nonce)
    return aesgcm.Seal(nonce, nonce, data, nil), nil
}

cosign sign --key cosign.key app.tar.gz
cosign verify app.tar.gz

cmd := exec.Command("cosign", "verify", "--key", key, file)
out, _ := cmd.CombinedOutput()

trivy image registry/app:v1.2

git clone https://github.com/spcent/deploylite.git
cd deploylite
docker-compose up -d

version: '3'
services:
  api:
    image: deploylite/api:latest
    ports: ["8080:8080"]
    depends_on: [postgres, redis, minio]
  scheduler:
    image: deploylite/scheduler:latest
  runner:
    image: deploylite/runner:latest
  postgres:
    image: postgres:15
    environment:
      POSTGRES_PASSWORD: example
  redis:
    image: redis:7
  minio:
    image: minio/minio
    command: server /data
    environment:
      MINIO_ACCESS_KEY: admin
      MINIO_SECRET_KEY: password
    ports:
      - "9000:9000"

global:
  imagePullPolicy: IfNotPresent

api:
  replicas: 3
  image: deploylite/api:v2.0
  resources:
    limits:
      cpu: 500m
      memory: 512Mi

scheduler:
  replicas: 2
  image: deploylite/scheduler:v2.0

runner:
  replicas: 4
  image: deploylite/runner:v2.0
  mode: docker

storage:
  endpoint: http://minio:9000
  bucket: artifacts

database:
  dsn: postgres://deploylite:pwd@postgres:5432/deploylite?sslmode=disable

redis:
  addr: redis:6379

promtail:
  scrape_configs:
    - job_name: deploylite
      static_configs:
        - targets:
            - localhost
          labels:
            app: deploylite
            __path__: /var/log/deploylite/*.log

apiVersion: autoscaling/v2
kind: HorizontalPodAutoscaler
metadata:
  name: deploylite-runner
spec:
  scaleTargetRef:
    kind: Deployment
    name: deploylite-runner
  minReplicas: 2
  maxReplicas: 10
  metrics:
  - type: Resource
    resource:
      name: cpu
      target:
        type: Utilization
        averageUtilization: 75

resource "aws_instance" "deploylite_api" {
  ami           = "ami-0c55b159cbfafe1f0"
  instance_type = "t3.medium"
  tags = {
    Name = "DeployLite-API"
  }
}

stages:
  - name: deploy-blue
    run: kubectl apply -f deployment-blue.yaml
  - name: switch-traffic
    run: kubectl patch svc app --patch '{"spec": {"selector": {"version": "blue"}}}'
  - name: remove-green
    run: kubectl delete deploy app-green

func TestParsePipelineYAML(t *testing.T) {
    yamlData := `
        name: test
        stages:
          - name: build
            steps:
              - run: go build
    `
    p, err := ParseYAML([]byte(yamlData))
    assert.NoError(t, err)
    assert.Equal(t, "test", p.Name)
    assert.Len(t, p.Stages, 1)
}

func TestCircularDependency(t *testing.T) {
    p := Pipeline{
        Stages: []Stage{
            {Name: "A", Needs: []string{"B"}},
            {Name: "B", Needs: []string{"A"}},
        },
    }
    _, err := BuildDAG(&p)
    assert.Error(t, err)
    assert.Contains(t, err.Error(), "circular")
}

func TestPipelineExecutionIntegration(t *testing.T) {
    ctx := context.Background()
    redisContainer, _ := testcontainers.GenericContainer(ctx, testcontainers.GenericContainerRequest{
        ContainerRequest: testcontainers.ContainerRequest{
            Image: "redis:7",
            ExposedPorts: []string{"6379/tcp"},
        },
        Started: true,
    })

    defer redisContainer.Terminate(ctx)

    engine := NewPipelineEngine(testConfig)
    err := engine.Run(ctx, samplePipeline)
    assert.NoError(t, err)
}

mockery --name=Storage --output=tests/mocks

type MockStorage struct {
    mock.Mock
}

func (m *MockStorage) Save(ctx context.Context, r io.Reader, key string) (int64, error) {
    args := m.Called(ctx, r, key)
    return args.Get(0).(int64), args.Error(1)
}

storage := new(MockStorage)
storage.On("Save", mock.Anything, mock.Anything, "file").Return(1024, nil)
svc := ArtifactService{Storage: storage}

import { test, expect } from '@playwright/test';

test('Create and Run Pipeline', async ({ page }) => {
  await page.goto('http://localhost:8080');
  await page.click('button#create-pipeline');
  await page.fill('input[name="name"]', 'demo');
  await page.click('button#save');
  await page.click('button#run');
  await expect(page.locator('.status')).toHaveText('success');
});

npx playwright test --reporter=html

echo "GET http://localhost:8080/api/v1/pipelines" | vegeta attack -duration=30s -rate=100 | vegeta report

import http from 'k6/http';
export let options = { vus: 50, duration: '1m' };
export default function () {
  http.get('http://localhost:8080/api/v1/pipelines');
}

go test ./... -coverprofile=coverage.out
go tool cover -html=coverage.out

- name: Check Coverage
  run: |
    go test ./... -cover | grep -E "coverage: (8[5-9]|9[0-9])%" || exit 1    

func TestPipelineRegression(t *testing.T) {
    got := ExecutePipeline(sample)
    want := LoadGolden("testdata/pipeline_output.golden")
    if diff := cmp.Diff(want, got); diff != "" {
        t.Errorf("Mismatch (-want +got):\n%s", diff)
    }
}

name: DeployLite QA
on:
  push:
    branches: [main]
jobs:
  test:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v3
      - name: Run Unit Tests
        run: go test ./tests/unit -v
      - name: Run Integration Tests
        run: go test ./tests/integration -v
      - name: Run E2E Tests
        run: npx playwright test

go test ./... -json > report.json
allure generate report.json --clean

task := redis.LPop("queue")
assign(task)

queueKey := fmt.Sprintf("queue:%s", tenantID)
task := redis.BLPop(ctx, 0, queueKey).Val()

hash := crc32.ChecksumIEEE([]byte(task.ProjectID))
runnerIndex := hash % len(runners)
assign(runners[runnerIndex])

tasks := redis.LRange("queue", 0, 9)
for _, t := range tasks {
    dispatch(t)
}
redis.LTrim("queue", 10, -1)

go func() {
    reportStatus(task.ID, "running")
}()

reportCh := make(chan Status, 100)
go worker(reportCh)

type Pool struct {
    tasks chan func()
    wg    sync.WaitGroup
}

func (p *Pool) Submit(task func()) {
    p.tasks <- task
}

func (p *Pool) worker() {
    for t := range p.tasks {
        t()
    }
}

docker run --name runner-slot-1 sleep infinity

docker exec runner-slot-1 sh -c "go build"

export GOGC=150

var bufPool = sync.Pool{New: func() interface{} {
    return new(bytes.Buffer)
}}

import "github.com/json-iterator/go"
var json = jsoniter.ConfigCompatibleWithStandardLibrary

var limiter = rate.NewLimiter(50, 100) // 每秒50请求
if !limiter.Allow() {
    http.Error(w, "Too Many Requests", 429)
}

func uploadParts(file io.Reader, parts int) {
    wg := sync.WaitGroup{}
    for i := 0; i < parts; i++ {
        wg.Add(1)
        go func() {
            defer wg.Done()
            uploadChunk()
        }()
    }
    wg.Wait()
}

func (c *Cache) GetArtifact(key string) (Artifact, error) {
    if val, ok := c.lru.Get(key); ok {
        return val.(Artifact), nil
    }
    data, err := redisClient.Get(ctx, key).Result()
    if err == nil {
        c.lru.Set(key, data)
    }
    return data, err
}

CREATE INDEX idx_pipeline_project_status ON pipelines(project_id, status);
CREATE INDEX idx_artifact_created_at ON artifacts(created_at DESC);

if readOnly {
    db = readReplica
} else {
    db = master
}

db.CreateInBatches(&tasks, 100)

SELECT * FROM pipelines WHERE id > ? ORDER BY id LIMIT 100;

app.Use(compress.New(compress.Config{
    Level: compress.LevelBestSpeed,
}))

batch := []Log{}
for log := range logCh {
    batch = append(batch, log)
    if len(batch) >= 100 {
        writeBatch(batch)
        batch = batch[:0]
    }
}

type Event struct {
    Type string
    Data string
}
eventCh := make(chan Event, 100)
go func() {
    for e := range eventCh {
        processEvent(e)
    }
}()

func BenchmarkScheduler(b *testing.B) {
    for i := 0; i < b.N; i++ {
        scheduler.Dispatch(task)
    }
}

go test -bench=. -benchmem > bench.txt
benchstat old.txt bench.txt

# Pseudocode
for task in tasks:
    runner = min(runners, key=lambda r: r.load + α*r.latency)
    assign(task, runner)

if lastAccess > 90d {
    moveToGlacier(file)
}