Conductor

High-Performance ETL Platform for Modern Data Engineering

Conductor is a powerful, enterprise-grade ETL (Extract, Transform, Load) platform designed for seamless data orchestration between databases, APIs, and file systems. Built with modern .NET and featuring a beautiful Svelte frontend, it delivers exceptional performance through advanced parallelization and channel-based processing.

✨ Features

🚀 Core Capabilities

High-Performance ETL: Channel-based producer-consumer pattern for maximum throughput
Multi-Database Support: PostgreSQL, MySQL, SQLite, SQL Server with connection pooling
RESTful API Integration: Native support for HTTP/HTTPS endpoints with pagination
Parallel Processing: Configurable parallel execution with intelligent resource management
Real-time Monitoring: Built-in job tracking, health checks, and performance metrics

🎨 Modern Interface

Intuitive Dashboard: Clean, responsive Svelte-based UI with Tailwind CSS
Job Management: Visual job scheduling, monitoring, and execution control
Data Preview: Interactive data exploration and validation tools
Real-time Updates: Live status updates and progress tracking

🔒 Enterprise Security

LDAP Authentication: Enterprise directory integration
Data Encryption: AES-256 encryption for sensitive connection strings
Network Security: IP filtering, CORS configuration, and HTTPS support
Role-based Access: Fine-grained permission system

📊 Observability

OpenTelemetry: Comprehensive distributed tracing
Structured Logging: Serilog with configurable log levels
Health Monitoring: Docker-ready health checks
Error Notifications: Webhook-based error alerting

🔧 Deployment Ready

Docker Support: Complete containerization with Docker Compose
Environment Configuration: Flexible configuration management
Scalability: Horizontal scaling capabilities
Cloud Ready: Kubernetes and cloud platform compatible

🏗️ Architecture

graph TB
    subgraph "Frontend Layer"
        UI[Svelte UI]
        API_CLIENT[API Client]
    end
    
    subgraph "API Layer"
        CONTROLLER[Controllers]
        MIDDLEWARE[Middleware]
        AUTH[Authentication]
    end
    
    subgraph "Business Logic"
        PIPELINE[Extraction Pipeline]
        SCHEDULER[Job Scheduler]
        TRANSFORMER[Data Transformer]
    end
    
    subgraph "Data Layer"
        REPO[Repositories]
        POOL[Connection Pool]
        MEMORY[Memory Manager]
    end
    
    subgraph "External Systems"
        DB[(Databases)]
        API_EXT[External APIs]
        FILES[File Systems]
    end
    
    UI --> API_CLIENT
    API_CLIENT --> CONTROLLER
    CONTROLLER --> MIDDLEWARE
    MIDDLEWARE --> AUTH
    CONTROLLER --> PIPELINE
    CONTROLLER --> SCHEDULER
    PIPELINE --> TRANSFORMER
    PIPELINE --> REPO
    REPO --> POOL
    POOL --> DB
    PIPELINE --> API_EXT
    PIPELINE --> FILES
    SCHEDULER --> PIPELINE
    MEMORY --> PIPELINE

Key Components

🎭 Extraction Pipeline: Core processing engine with parallel execution
🗄️ Connection Pool Manager: Efficient database connection management
🧠 Memory Manager: Intelligent DataTable lifecycle management
📋 Job Tracker: Comprehensive job scheduling and monitoring
🔄 Data Transformers: Flexible data transformation capabilities
🌐 HTTP Exchange: RESTful API integration layer

🚀 Quick Start

Get Conductor running in under 5 minutes:

# Clone the repository
git clone https://github.com/leonardomb1/Conductor.git
cd Conductor

dotnet publish -c Release linux-x64 (or the platform you're using)

cd build/publish/
./Conductor -fM ../../.env

That's it! Conductor is now running as an API and ready to process your data.

The docker setup is recommend as it has a web interface aswell:

# Clone the repository
git clone https://github.com/leonardomb1/Conductor.git
cd Conductor

docker compose up -d

open http://localhost:3000

📖 Installation

Prerequisites

.NET SDK 9.0+ - Download here
Node.js 18+ - Download here
Docker & Docker Compose - Install Docker
PostgreSQL (optional - included in Docker setup)

Option 1: Docker Deployment (Recommended)

# Clone and configure
git clone https://github.com/leonardomb1/Conductor.git
cd Conductor

# Copy and configure environment
cp .env.example .env
# Edit .env with your settings

# Deploy with Docker Compose
docker compose up -d

# Check health status
docker compose ps

Option 2: Local Development

# Backend setup
cd src/Backend
dotnet restore
dotnet build

# Frontend setup  
cd ../Frontend
npm install
npm run build

# Configure environment
cp .env.example .env
# Edit .env with your database settings

# Run migrations
dotnet run -- --migrate-init-file

# Start the application
dotnet run -- --file

Option 3: Production Binary

# Download latest release
wget https://github.com/leonardomb1/Conductor/releases/latest/download/conductor-linux-x64.tar.gz

# Extract and run
tar -xzf conductor-linux-x64.tar.gz
cd conductor

# Configure environment
cp .env.example .env
# Edit .env with your settings

# Run migrations and start
./Conductor --migrate-init-file

⚙️ Configuration

Environment Variables

Create a .env file in the project root, you can use the .env.example as a basis and edit for your use.

Database Support

Conductor supports multiple database systems:

Database	Connection String Example
PostgreSQL	`Server=localhost;Port=5432;Database=mydb;User Id=user;Password=pass;`
MySQL	`Server=localhost;Port=3306;Database=mydb;Uid=user;Pwd=pass;`
SQLite	`Data Source=app.db;Mode=ReadWriteCreate;`
SQL Server	`Server=localhost;Database=mydb;User Id=user;Password=pass;`

🎯 Usage

1. Creating Data Sources

Database Origin

{
  "originName": "Production Database",
  "originDbType": "PostgreSQL",
  "originConStr": "Server=prod-db;Port=5432;Database=sales;User Id=reader;Password=***;",
  "": "Server=prod-db;Port=5432;Database=sales;User Id=reader;Password=***;",
  "originTimeZoneOffSet": 0
}

2. Configuring Extractions

{
  "name": "Daily Sales Extract",
  "originId": 1,
  "destinationId": 2,
  "scheduleId": 1,
  "overrideQuery": "SELECT * FROM sales WHERE created_date >= '2025-01-01'"
}

3. Running Extractions

Via API

# Execute transfer (source to destination)
curl -X POST "http://localhost:8080/api/extractions/transfer?scheduleId=1" \
  -H "Authorization: Key your-api-key"

# Execute pull (source to CSV)
curl -X POST "http://localhost:8080/api/extractions/pull?scheduleId=1" \
  -H "Authorization: Key your-api-key"

# Fetch data preview
curl -X GET "http://localhost:8080/api/extractions/fetch?id=1&page=1" \
  -H "Authorization: Key your-api-key"

Via Frontend

Navigate to Extractions in the dashboard
Select your extraction configuration
Click Execute Transfer or Pull Data
Monitor progress in the Jobs section

4. Scheduling Jobs

To be done in the future. Recommended to use a scheduler such as Airflow for this.

🔌 API Reference

Authentication

When Authentication is activated in the environment variable, all routes, barring ssologin and login routes, require authentication through 'Authorization' header prefixed with either: 1. Key, when using API master key; 2. Bearer, JWT token gained by succesfully logging in the login or ssologin routes; or 3. Basic, with a Base64 encoded user and password.

It is recomended to always use Bearer tokens whenever possible.

Authorization: {Basic|Bearer|Key} your_secret

Core Endpoints

Extractions

GET /api/extractions - List all extractions
POST /api/extractions - Create new extraction
GET /api/extractions/{id} - Get extraction details
PUT /api/extractions/{id} - Update extraction
DELETE /api/extractions/{id} - Delete extraction
POST /api/extractions/transfer - Execute transfer job
POST /api/extractions/pull - Execute pull job
GET /api/extractions/fetch - Preview data

Jobs

GET /api/jobs - List all jobs
GET /api/jobs/{id} - Get job details
DELETE /api/jobs/{id} - Cancel job

Origins & Destinations

GET /api/origins - List data sources
POST /api/origins - Create data source
GET /api/destinations - List destinations
POST /api/destinations - Create destination

Schedules

GET /api/schedules - List schedules
POST /api/schedules - Create schedule
PUT /api/schedules/{id} - Update schedule

Query Parameters

Filtering

name - Filter by exact name
contains - Filter by partial name match
scheduleId - Filter by schedule ID
originId - Filter by origin ID
take - Limit results

Pagination

page - Page number (1-based)
pageSize - Results per page

Response Format

{
  "statusCode": 200,
  "message": "Success",
  "content": [...],
  "page": 1,
  "success": true
}

🌊 Data Flow

Transfer Process

Source Query - Execute query against origin database/API
Data Extraction - Fetch data using parallel processing
Transformation - Apply any configured transformations
Destination Load - Insert/update data in destination
Monitoring - Track progress and handle errors

Pull Process

Source Query - Execute query against origin
Data Extraction - Fetch data in batches
CSV Generation - Convert to CSV format
File Export - Save to configured output directory

Parallel Processing

Producer-Consumer Pattern - Separate data extraction and loading
Channel-based Communication - High-performance data pipeline
Configurable Parallelism - Optimize for your infrastructure
Memory Management - Automatic cleanup and optimization

🔧 Advanced Configuration

Connection Pooling

# Connection pool settings
CONNECTION_POOL_MAX_SIZE=50
CONNECTION_POOL_MIN_SIZE=5
CONNECTION_IDLE_TIMEOUT_MINUTES=30

Memory Management

# DataTable memory management
DATATABLE_LIFETIME_MINUTES=30
DATATABLE_CLEANUP_INTERVAL_MINUTES=5
MEMORY_PRESSURE_THRESHOLD_GB=4
GC_CHECK_INTERVAL_MINUTES=1

Performance Tuning

# Parallel processing
MAX_DEGREE_PARALLEL=20
MAX_CONSUMER_FETCH=200
MAX_PRODUCER_LINECOUNT=100000

# Timeouts
QUERY_TIMEOUT_SEC=1800
CONNECTION_TIMEOUT_MIN=15

Custom Transformations

// Example transformation script
public class CustomTransformer : IDataTransformer
{
    public async Task<Result<DataTable>> TransformAsync(DataTable input, string script)
    {
        // Your transformation logic here
        return Result.Ok(transformedData);
    }
}

📊 Monitoring & Observability

Health Checks

# Check API health
curl http://localhost:8080/api/health

Metrics & Tracing

OpenTelemetry Integration - Distributed tracing
Prometheus Metrics - Performance monitoring
Structured Logging - Comprehensive audit trail
Custom Dashboards - Grafana-compatible metrics

Log Levels

# Configure logging
LOGGING_LEVEL_DEBUG=false
DEVELOPMENT_MODE=false
SEND_WEBHOOK_ON_ERROR=true

Error Handling

Automatic Retries - Configurable retry policies
Dead Letter Queues - Failed message handling
Webhook Notifications - Real-time error alerts
Graceful Degradation - Partial failure handling

🔐 Security

Authentication Methods

API Key Authentication - Simple key-based access
LDAP Integration - Enterprise directory services
Role-based Access Control - Planned for the future

Data Protection

Encryption at Rest - AES-256 encryption for sensitive data
Encryption in Transit - TLS/SSL for all communications
Connection String Encryption - Automatic credential protection
IP Whitelisting - Network-level access control

Security Best Practices

# Generate secure encryption key
openssl rand -base64 32

# Configure secure headers
USE_HTTPS=true
VERIFY_CORS=true
VERIFY_TCP=true

# Set up IP filtering
ALLOWED_IP_ADDRESSES="10.0.0.0/8|192.168.0.0/16"

🐳 Docker Deployment

Docker Compose (Recommended)

# docker-compose.yml
version: '3.8'

services:
  conductor-api:
    image: conductor-api:latest
    environment:
      - CONNECTION_STRING=Server=database;Port=5432;Database=ConductorDb;User Id=conductor;Password=password;
    depends_on:
      - database
    ports:
      - "8080:8080"

  conductor-frontend:
    image: conductor-frontend:latest
    environment:
      - API_HOST=conductor-api:8080
    ports:
      - "3000:3000"

  database:
    image: postgres:15
    environment:
      - POSTGRES_DB=ConductorDb
      - POSTGRES_USER=conductor
      - POSTGRES_PASSWORD=password
    volumes:
      - postgres_data:/var/lib/postgresql/data
    ports:
      - "5432:5432"

volumes:
  postgres_data:

Kubernetes Deployment

# kubernetes-deployment.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: conductor-api
spec:
  replicas: 3
  selector:
    matchLabels:
      app: conductor-api
  template:
    metadata:
      labels:
        app: conductor-api
    spec:
      containers:
      - name: conductor-api
        image: conductor-api:latest
        ports:
        - containerPort: 8080
        env:
        - name: CONNECTION_STRING
          valueFrom:
            secretKeyRef:
              name: conductor-secrets
              key: connection-string

🧪 Development

Prerequisites

.NET 9.0 SDK
Node.js 18+
Docker Desktop
PostgreSQL (optional)

Project Structure

Conductor/
├── src/
│   ├── Backend/
│   │   ├── Controller/          # API controllers
│   │   ├── Service/             # Business logic
│   │   ├── Repository/          # Data access
│   │   ├── Model/               # Entity models
│   │   ├── Middleware/          # Custom middleware
│   │   └── Shared/              # Shared utilities
│   └── Frontend/
│       ├── src/
│       │   ├── routes/          # Svelte routes
│       │   ├── lib/             # Shared components
│       │   └── app.html         # Main template
│       └── static/              # Static assets
├── docker/                      # Docker configurations
├── .github/                     # GitHub Actions
└── docs/                        # Documentation

🤝 Contributing

We welcome contributions!

Development Workflow

Fork the repository
Create a feature branch (git checkout -b feature/amazing-feature)
Commit your changes (git commit -m 'Add amazing feature')
Push to the branch (git push origin feature/amazing-feature)
Open a Pull Request

Code Standards

Follow .NET coding conventions
Use Conventional Commits for commit messages
Update documentation for new features

Reporting Issues

Use the issue tracker
Include reproduction steps and environment details
Label issues appropriately (bug, enhancement, question)

📝 License

This project is licensed under the MIT License - see the LICENSE file for details.

🙏 Acknowledgments

Microsoft .NET Team for the excellent .NET platform
Svelte Team for the amazing frontend framework
PostgreSQL Community for the robust database
Docker Team for containerization platform
All contributors and users of this project

⭐ Star this repository if you find it useful!

Name		Name	Last commit message	Last commit date
Latest commit History 201 Commits
.config		.config
.github/workflows		.github/workflows
docker		docker
src		src
.dockerignore		.dockerignore
.env.example		.env.example
.gitignore		.gitignore
Conductor.sln		Conductor.sln
LICENSE		LICENSE
README.md		README.md
docker-compose.yml		docker-compose.yml
migrate.sh		migrate.sh

Folders and files

Latest commit

History

Repository files navigation

Conductor

📋 Table of Contents

✨ Features

🚀 Core Capabilities

🎨 Modern Interface

🔒 Enterprise Security

📊 Observability

🔧 Deployment Ready

🏗️ Architecture

Key Components

🚀 Quick Start

📖 Installation

Prerequisites

Option 1: Docker Deployment (Recommended)

Option 2: Local Development

Option 3: Production Binary

⚙️ Configuration

Environment Variables

Database Support

🎯 Usage

1. Creating Data Sources

Database Origin

2. Configuring Extractions

3. Running Extractions

Via API

Via Frontend

4. Scheduling Jobs

🔌 API Reference

Authentication

Core Endpoints

Extractions

Jobs

Origins & Destinations

Schedules

Query Parameters

Filtering

Pagination

Response Format

🌊 Data Flow

Transfer Process

Pull Process

Parallel Processing

🔧 Advanced Configuration

Connection Pooling

Memory Management

Performance Tuning

Custom Transformations

📊 Monitoring & Observability

Health Checks

Metrics & Tracing

Log Levels

Error Handling

🔐 Security

Authentication Methods

Data Protection

Security Best Practices

🐳 Docker Deployment

Docker Compose (Recommended)

Kubernetes Deployment

🧪 Development

Prerequisites

Project Structure

🤝 Contributing

Development Workflow

Code Standards

Reporting Issues

📝 License

🙏 Acknowledgments

About

Topics

Resources

Stars

Watchers

Forks

Releases

Packages