How Harbor's Replication Controller Works: Policy and Execution Architecture

Harbor's Replication Controller is a global orchestrator (replication.Ctl) that manages replication policies and executes cross-registry synchronization through a bounded worker pool, delegating to copy or deletion flows that create jobservice tasks.

The replication subsystem in goharbor/harbor centers on a central controller implementing the replication.Controller interface. This component coordinates artifact synchronization between registries by managing policy lifecycles, spawning execution goroutines with isolated database contexts, and maintaining state through Harbor's task management system. Understanding how Harbor's Replication Controller works requires examining its worker pool architecture, flow selection logic, and task creation mechanisms.

Architecture of the Replication Controller

The controller is defined in src/controller/replication/execution.go and exposes CRUD operations for replication policies alongside lifecycle methods for executions and tasks. It acts as the primary entry point for all replication operations, forwarding policy management requests to the persistence layer (replication.Mgr) while handling execution orchestration directly.

All database interactions utilize the ORM creator (orm.Crt) to ensure each goroutine operates within its own transactional context. This design prevents connection leaks and guarantees that concurrent replication workers remain isolated from one another.

Policy Management Operations

Policy management methods including PolicyCount, ListPolicies, CreatePolicy, UpdatePolicy, and DeletePolicy handle the static configuration defining what to replicate. These methods accept context and query parameters, forwarding requests to the underlying manager to interact with the database. While these operations define source and destination registries, resource filters, and trigger modes, they do not initiate the actual data transfer.

Starting a Replication Execution

The Ctl.Start method in src/controller/replication/execution.go (lines 1-80) initiates the active synchronization process. It accepts a policy object, an optional specific resource, and a trigger type (e.g., "manual", "schedule").

When invoked, the method first creates an execution record via task.ExecMgr.Create to track the overall job. It then validates concurrent execution constraints—if the policy allows only one active run and another execution is already in progress, the request is rejected. Upon validation, the controller acquires a worker from the pool, creates a new ORM context, and launches a goroutine that invokes flow.NewController().Start.

Concurrency Control and Worker Pool

To prevent resource exhaustion, the controller maintains a worker pool (wp *lib.WorkerPool) as a struct field with a default capacity of 10 concurrent goroutines. This bounded pool ensures that Harbor does not overwhelm the jobservice or target registries with unlimited concurrent replication operations. Each spawned goroutine operates independently with its own database transaction context.

Flow Selection: Copy vs Deletion

The flow controller (src/controller/replication/flow/controller.go) determines the execution path based on the resource's deletion state. If the supplied resource is marked as Deleted, the controller initiates the deletion flow; otherwise, it routes to the copy flow. This logic ensures that artifact removal and synchronization follow distinct code paths while sharing common infrastructure.

Inside the Copy Flow

The copy flow implementation in src/controller/replication/flow/copy.go handles artifact synchronization through four discrete steps:

1. Initialize adapters for the source and destination registries
2. Fetch resources from the source registry if none were explicitly supplied in the request
3. Assemble resource pairs respecting policy filters, tags, and labels
4. Create tasks for each resource pair via taskMgr.Create

Each task registers a jobservice job of type ReplicationVendorType containing serialized parameters including source and destination resource JSON, speed limits, and chunking preferences. The jobservice subsequently processes these tasks to perform the actual image pull and push operations.

Inside the Deletion Flow

The deletion flow in src/controller/replication/flow/deletion.go follows a similar architectural pattern but handles removal of resources from the target registry. When executed, it ensures that artifacts deleted from the source are purged from the destination when the policy requires synchronized deletion, maintaining consistency between registry states.

Monitoring and Stopping Executions

Replication executions can be monitored through Ctl.ListExecutions and Ctl.ListTasks, which query the execution and task managers (task.Mgr) to return current status, start times, and completion metrics. Individual task logs are retrieved via Ctl.GetTaskLog, which streams data directly from the jobservice storage backend.

To halt an active replication, the Ctl.Stop method (lines 89-97 of execution.go) retrieves the execution record and delegates to execMgr.Stop. This signals the execution manager to cancel the operation, which in turn requests the jobservice to terminate any running tasks associated with that execution ID.

Practical Code Examples

Starting a Replication Programmatically

import (
    "context"
    "github.com/goharbor/harbor/src/controller/replication"
    "github.com/goharbor/harbor/src/pkg/reg/model"
)

func startReplication(ctx context.Context, policyID int64) (int64, error) {
    // fetch the policy (error handling omitted for brevity)
    pol, _ := replication.Ctl.GetPolicy(ctx, policyID)

    // optional: limit to a single resource, otherwise nil means “all”
    var res *model.Resource = nil

    // trigger can be "manual", "schedule", etc.
    execID, err := replication.Ctl.Start(ctx, pol, res, "manual")
    if err != nil {
        return 0, err
    }
    return execID, nil
}

Relevant source: Ctl.Start implementation – lines 1-80 of execution.go.

Stopping an Ongoing Execution

func stopReplication(ctx context.Context, executionID int64) error {
    return replication.Ctl.Stop(ctx, executionID)
}

Relevant source: Ctl.Stop – lines 89-97 of execution.go.

Listing Tasks and Retrieving Logs

func listTasks(ctx context.Context, execID int64) ([]*replication.Task, error) {
    q := &q.Query{Keywords: map[string]any{"ExecutionID": execID}}
    return replication.Ctl.ListTasks(ctx, q)
}

func getTaskLog(ctx context.Context, taskID int64) ([]byte, error) {
    return replication.Ctl.GetTaskLog(ctx, taskID)
}

Relevant source: ListTasks and GetTaskLog – lines 48-73 of execution.go.

Task Creation in the Copy Flow

job := &task.Job{
    Name: job.ReplicationVendorType,
    Metadata: &job.Metadata{JobKind: job.KindGeneric},
    Parameters: map[string]any{
        "src_resource":  string(srcJSON),
        "dst_resource":  string(dstJSON),
        "speed":         speed,
        "copy_by_chunk": copyByChunk,
    },
}
taskMgr.Create(ctx, executionID, job, map[string]any{
    "operation":            "copy",
    "resource_type":        string(src.Type),
    "source_resource":      getResourceName(src),
    "destination_resource": getResourceName(dst),
    "references":           getResourceReferences(dst),
})

Relevant source: copyFlow.createTasks – lines 34-55 of copy.go.

Summary

• Global Orchestration: The replication.Ctl controller in src/controller/replication/execution.go implements the replication.Controller interface to manage policies, executions, and tasks.
• Worker Pool: A bounded pool of 10 goroutines (configurable via lib.WorkerPool) limits concurrent replication jobs to prevent resource exhaustion.
• Flow Selection: The flow controller (src/controller/replication/flow/controller.go) routes execution to either copy or deletion logic based on the resource deletion state.
• Task Creation: Both flows create individual tasks via taskMgr.Create, registering jobs of type ReplicationVendorType with the Harbor jobservice for actual data transfer.
• Transaction Isolation: Each replication goroutine creates its own ORM context using orm.Crt to ensure database transaction isolation.
• Lifecycle Management: Methods like Ctl.Start and Ctl.Stop provide full lifecycle control, while GetTaskLog enables real-time monitoring of replication progress.

Frequently Asked Questions

What is the difference between a replication policy and an execution?

A replication policy is the static configuration stored in Harbor that defines source and destination registries, resource filters, and trigger rules. An execution is a runtime instance of that policy—essentially a single "run" that tracks the status of the replication job, including start time, end time, and overall success or failure. While the policy defines what to replicate and when, the execution tracks the actual progress of the synchronization.

How does Harbor prevent concurrent replication runs for the same policy?

The Ctl.Start method checks the policy's concurrent execution constraints before launching a new goroutine. If the policy is configured to allow only one active execution (the default for many policies), the controller queries existing executions and rejects new requests if any are currently running. This check happens after creating the execution record but before acquiring a worker pool slot, ensuring database consistency while preventing resource contention.

What happens when a replication execution is stopped?

When Ctl.Stop is called, the controller retrieves the execution record and delegates to execMgr.Stop. This updates the execution status and signals the jobservice to cancel any running tasks associated with that execution ID. The jobservice then terminates the underlying replication jobs, which may leave partially transferred artifacts in the destination registry depending on when the cancellation occurred.

How are replication tasks distributed to workers?

The controller uses a worker pool (lib.WorkerPool) to manage concurrency. When Ctl.Start is invoked, it submits the replication flow to the pool, which ensures no more than 10 (by default) replication goroutines execute simultaneously. Each worker operates independently with its own ORM context, fetching resources, creating tasks, and monitoring jobservice completion without blocking other replication policies.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}