Compare commits

..

43 Commits

Author SHA1 Message Date
charlesgauthereau d7740e05df chore(release): 1.0.3-rc.3 2026-01-28 12:25:34 +01:00
charlesgauthereau 2688ed7760 fix: docker.yml 2026-01-28 12:25:33 +01:00
charlesgauthereau a2067c063d chore(release): 1.0.3-rc.2 2026-01-28 11:55:57 +01:00
charlesgauthereau d4b597ed27 chore(release): 1.0.3-rc.2 2026-01-28 11:55:11 +01:00
charlesgauthereau 927d8d3610 fix: docker.yml 2026-01-28 11:55:03 +01:00
charlesgauthereau 7576b9bdfb chore(release): 1.0.3-rc.1 2026-01-28 10:55:41 +01:00
charlesgauthereau 1635a74c9f fix: mongo docker configuration. 2026-01-28 10:53:15 +01:00
Charles GTE e41a7a7c5b Merge pull request #4 from Portabase/fix/mongo-folder
fix: mongo domain folder name
2026-01-24 19:15:27 +01:00
charlesgauthereau 2b7c7a3bd1 fix: mongo domain folder name 2026-01-24 19:14:54 +01:00
charlesgauthereau e7f072c9c8 fix: .gitignore 2026-01-22 10:50:42 +01:00
charlesgauthereau d9c5cb7a09 fix: .gitignore 2026-01-22 10:49:03 +01:00
charlesgauthereau b45b227c91 fix: .gitignore 2026-01-22 10:48:39 +01:00
charlesgauthereau 8032639da6 chore(release): 1.0.2 2026-01-22 10:47:46 +01:00
Charles GTE c418164b48 Merge pull request #3 from Portabase/feat/arm-architecture
feat/arm-architecture
2026-01-22 10:46:40 +01:00
Charles GTE b4095ca51a chore: README.md 2026-01-22 10:42:20 +01:00
charlesgauthereau 48b1cb07f5 chore(release): 1.0.2-rc.1 2026-01-21 21:48:07 +01:00
charlesgauthereau a21e9b385c feat: Docker image for ARM architecture 2026-01-21 21:47:06 +01:00
Charles GTE 8f89830ccc chore: Update README.md 2026-01-20 23:03:19 +01:00
Charles GTE 3480e332f2 chore: README.md 2026-01-20 23:00:21 +01:00
Charles GTE 0389e24af2 Merge pull request #2 from Portabase/dev
fix: version_str when ping server.
2026-01-20 22:46:36 +01:00
charlesgauthereau c68d4ee5f4 fix: version_str when ping server. 2026-01-20 22:45:55 +01:00
charlesgauthereau a1eaa2a671 chore(release): 1.0.1 2026-01-20 22:44:46 +01:00
Charles GTE 79ed82dd15 Merge pull request #1 from Portabase/feat/add-mongodb
feat/add-mongodb
2026-01-20 22:43:16 +01:00
charlesgauthereau da07f03878 feat: Adding MongoDB support 2026-01-20 22:42:22 +01:00
charlesgauthereau e867e857ee feat: MongoDB provider 2026-01-18 21:19:55 +01:00
charlesgauthereau f36b816941 chore(release): 1.0.1-rc.1 2026-01-12 11:53:09 +01:00
charlesgauthereau a11b81bfdd fix: patch timezone error. 2026-01-12 11:52:57 +01:00
charlesgauthereau 1229dc2769 fix: Cargo.lock 2026-01-09 19:18:04 +01:00
charlesgauthereau 0d771c7ea4 chore(release): 1.0.0 2026-01-09 19:17:22 +01:00
charlesgauthereau ee3924d070 fix: refactoring mysql side and tested the method. Add logging with config from env 2026-01-09 19:16:50 +01:00
charlesgauthereau 20e3facb58 chore(release): 1.0.0-rc.1 2026-01-09 19:05:01 +01:00
charlesgauthereau 09991d3c48 fix: refactoring mysql side and tested the method. Add logging with config from env 2026-01-09 19:04:37 +01:00
Charles GTE 8a2eb4638a Delete .DS_Store 2026-01-08 23:23:27 +01:00
charlesgauthereau f7af1d5db3 fix: gitignore 2026-01-08 23:23:03 +01:00
charlesgauthereau 8973e3a17e fix: Postgres system: add a lock system to avoid concurrency problems. 2026-01-08 23:19:58 +01:00
charlesgauthereau c69ab2a266 fix: Cargo.lock 2026-01-08 09:52:43 +01:00
charlesgauthereau 7fc4e59a4c chore(release): 0.1.2-rc.1 2026-01-08 09:52:24 +01:00
charlesgauthereau 766abe697a fix: postgres.rs methods and add tokio-postgres 2026-01-08 09:52:05 +01:00
charlesgauthereau a279993fa5 chore: Cargo.lock 2026-01-07 22:31:49 +01:00
charlesgauthereau bb3f254e6c chore(release): 0.1.1 2026-01-07 22:31:20 +01:00
charlesgauthereau 5d1841cdfe fix: psql error for ping method in postgres 2026-01-07 22:30:46 +01:00
charlesgauthereau 942b0e421e chore(release): 0.1.1-rc.8 2026-01-07 22:22:55 +01:00
charlesgauthereau 5daa635c8c fix: psql error for ping method in postgres 2026-01-07 22:22:45 +01:00
55 changed files with 2927 additions and 933 deletions
+97 -13
View File
@@ -26,19 +26,26 @@ on:
required: true
jobs:
build-and-push:
build-amd64:
runs-on: ubuntu-latest
permissions:
packages: write
contents: read
env:
ARCH: amd64
steps:
- name: Check out the repo
- name: Checkout repo
uses: actions/checkout@v4
with:
fetch-depth: 0
- name: Log in to Docker Hub
uses: docker/login-action@f4ef78c080cd8ba55a85445d5b36e214a81df20a
- name: Set up QEMU
uses: docker/setup-qemu-action@v3
- name: Set up Buildx
uses: docker/setup-buildx-action@v3
with:
install: true
- name: Docker login
uses: docker/login-action@v2
with:
username: ${{ secrets.DOCKER_USERNAME }}
password: ${{ secrets.DOCKER_PASSWORD }}
@@ -47,17 +54,94 @@ jobs:
id: set-tags
run: |
REF_NAME=${GITHUB_REF#refs/tags/}
TAGS="${{ inputs.image_name }}:$REF_NAME"
if [[ "${{ inputs.add_latest }}" == "true" ]]; then
TAGS="$TAGS,${{ inputs.image_name }}:latest"
ARCH_TAG="${IMAGE_NAME}:$REF_NAME-${ARCH}"
TAGS="$ARCH_TAG"
if [ "${ADD_LATEST}" = "true" ]; then
TAGS="$TAGS,${IMAGE_NAME}:latest-${ARCH}"
fi
echo "tags=$TAGS" >> $GITHUB_OUTPUT
- name: Build and push Docker image
- name: Build and push amd64
uses: docker/build-push-action@v6
with:
context: .
file: ${{ inputs.dockerfile }}
file: ${{ env.DOCKERFILE }}
platforms: linux/amd64
push: true
tags: ${{ steps.set-tags.outputs.tags }}
target: ${{ inputs.target }}
target: ${{ env.TARGET }}
cache-from: type=registry,ref=${IMAGE_NAME}:buildcache
cache-to: type=registry,ref=${IMAGE_NAME}:buildcache,mode=max
build-arm64:
runs-on: ubuntu-latest
env:
ARCH: arm64
steps:
- name: Checkout repo
uses: actions/checkout@v4
with:
fetch-depth: 0
- name: Set up QEMU
uses: docker/setup-qemu-action@v3
- name: Set up Buildx
uses: docker/setup-buildx-action@v3
with:
install: true
- name: Docker login
uses: docker/login-action@v2
with:
username: ${{ secrets.DOCKER_USERNAME }}
password: ${{ secrets.DOCKER_PASSWORD }}
- name: Set tags
id: set-tags
run: |
REF_NAME=${GITHUB_REF#refs/tags/}
ARCH_TAG="${IMAGE_NAME}:$REF_NAME-${ARCH}"
TAGS="$ARCH_TAG"
if [ "${ADD_LATEST}" = "true" ]; then
TAGS="$TAGS,${IMAGE_NAME}:latest-${ARCH}"
fi
echo "tags=$TAGS" >> $GITHUB_OUTPUT
- name: Build and push arm64
uses: docker/build-push-action@v6
with:
context: .
file: ${{ env.DOCKERFILE }}
platforms: linux/arm64
push: true
tags: ${{ steps.set-tags.outputs.tags }}
target: ${{ env.TARGET }}
cache-from: type=registry,ref=${IMAGE_NAME}:buildcache
cache-to: type=registry,ref=${IMAGE_NAME}:buildcache,mode=max
manifest:
needs: [ build-amd64, build-arm64 ]
runs-on: ubuntu-latest
steps:
- name: Docker login
uses: docker/login-action@v2
with:
username: ${{ secrets.DOCKER_USERNAME }}
password: ${{ secrets.DOCKER_PASSWORD }}
- name: Create and push multi-arch manifest
run: |
REF_NAME=${GITHUB_REF#refs/tags/}
docker manifest create ${IMAGE_NAME}:$REF_NAME \
${IMAGE_NAME}:$REF_NAME-amd64 \
${IMAGE_NAME}:$REF_NAME-arm64
if [ "${ADD_LATEST}" = "true" ]; then
docker manifest create ${IMAGE_NAME}:latest \
${IMAGE_NAME}:latest-amd64 \
${IMAGE_NAME}:latest-arm64
docker manifest push ${IMAGE_NAME}:latest
fi
docker manifest push ${IMAGE_NAME}:$REF_NAME
+3
View File
@@ -1,3 +1,6 @@
/target
.idea
/src/data/
.DS_Store
.env
+2 -2
View File
@@ -22,5 +22,5 @@ keywords:
- self-hosted
- portabase
license: Apache-2.0
version: 0.1.1-rc.7
date-released: "2026-01-07"
version: 1.0.3-rc.3
date-released: "2026-01-28"
Generated
+1279 -98
View File
File diff suppressed because it is too large Load Diff
+8 -2
View File
@@ -1,6 +1,6 @@
[package]
name = "portabase-agent"
version = "0.1.1-rc.7"
version = "1.0.3-rc.3"
edition = "2024"
[dependencies]
@@ -10,7 +10,7 @@ chrono = { version = "0.4", features = ["serde"] }
serde = { version = "1.0", features = ["derive"] }
serde_json = "1.0"
tracing = "0.1"
tracing-subscriber = "0.3"
tracing-subscriber = { version = "0.3", features = ["env-filter", "fmt", "local-time", "chrono"] }
dotenvy = "0.15"
once_cell = "1.17"
base64 = "0.22.1"
@@ -26,6 +26,12 @@ openssl = "0.10.75"
hex = "0.4.3"
flate2 = "1.1.5"
tar = "0.4.44"
tokio-postgres = "0.7.15"
futures = "0.3.31"
tracing-log = "0.2.0"
tracing-appender = "0.2.4"
time = { version = "0.3.44", features = ["macros"] }
mongodb = "3.5.0"
[[bin]]
name = "app"
+29
View File
@@ -0,0 +1,29 @@
include .env
export $(shell sed 's/=.*//' .env)
.PHONY: seed-mongo seed-mysql seed-postgres
seed-mongo:
@echo "Seeding MongoDB..."
bash ./scripts/mongo/seed-mongo.sh
seed-mongo-auth:
@echo "Seeding MongoDB with auth..."
bash ./scripts/mongo/seed-mongo.sh auth
seed-mysql:
@echo "Seeding MySQL..."
mysql -h 127.0.0.1 -P "$$MYSQL_PORT" -u "$$MYSQL_USER" -p"$$MYSQL_PASSWORD" "$$MYSQL_DB" < ./scripts/mysql/seed-mysql.sql
seed-postgres:
@echo "Seeding Postgres..."
docker exec -i -e PGPASSWORD=$$PG_PASSWORD $$PG_CONTAINER \
psql -U $$PG_USER -d $$PG_DB < ./scripts/postgres/seed.sql
seed-postgres-1gb:
@echo "Seeding Postgres..."
docker exec -i -e PGPASSWORD=$$PG_PASSWORD $$PG_CONTAINER \
psql -U $$PG_USER -d $$PG_DB < ./scripts/postgres/seed-1gb.sql
seed-all: seed-mongo seed-mysql seed-postgres seed-postgres-1gb
+6 -4
View File
@@ -12,13 +12,15 @@
</p>
[![License: Apache](https://img.shields.io/badge/License-apache-yellow.svg)](LICENSE)
[![Docker Pulls](https://img.shields.io/docker/pulls/solucetechnologies/agent-portabase-rust?color=brightgreen)](https://hub.docker.com/r/solucetechnologies/agent-portabase)
[![Docker Pulls](https://img.shields.io/docker/pulls/portabase/agent?color=brightgreen)](https://hub.docker.com/r/portabase/agent)
[![Platform](https://img.shields.io/badge/platform-linux%20%7C%20macos%20%7C%20windows-lightgrey)](https://github.com/Portabase/portabase)
[![Support Portabase](https://img.shields.io/badge/Support-Portabase-orange)](https://www.buymeacoffee.com/portabase)
[![PostgreSQL](https://img.shields.io/badge/python-3.13-blue.svg)](https://www.python.org/downloads/release/python-3120/)
[![PostgreSQL](https://img.shields.io/badge/PostgreSQL-336791?logo=postgresql&logoColor=white)](https://www.postgresql.org/)
[![MySQL](https://img.shields.io/badge/MySQL-4479A1?logo=mysql&logoColor=white)](https://www.mysql.com/)
[![MariaDB](https://img.shields.io/badge/MariaDB-003545?logo=mariadb&logoColor=white)](https://mariadb.org/)
[![MongoDB](https://img.shields.io/badge/-MongoDB-13aa52?logo=mongodb&logoColor=white)](https://www.mongodb.com/)
[![Self Hosted](https://img.shields.io/badge/self--hosted-yes-brightgreen)](https://github.com/Portabase/portabase)
[![Open Source](https://img.shields.io/badge/open%20source-❤️-red)](https://github.com/Portabase/portabase)
@@ -53,9 +55,9 @@ You have 4 ways to install Portabase Agent:
## Contributors
<a href="https://github.com/Portabase/agent-portabase/graphs/contributors">
<img src="https://contrib.rocks/image?repo=Portabase/agent-portabase" alt="contrib.rocks image" />
</a>
[![Contributors](https://contrib.rocks/image?repo=Portabase/agent-rust)](https://github.com/Portabase/agent-rust/graphs/contributors)
[!["Support Portabase"](https://www.buymeacoffee.com/assets/img/custom_images/orange_img.png)](https://www.buymeacoffee.com/portabase)
## License
BIN
View File
Binary file not shown.
+44
View File
@@ -0,0 +1,44 @@
{
"databases": [
{
"name": "Test database 1 - PostgreSQL",
"database": "devdb",
"type": "postgresql",
"username": "devuser",
"password": "changeme",
"port": 5432,
"host": "db-postgres",
"generated_id": "16678159-ff7e-4c97-8c83-0adeff214681"
},
{
"name": "Test database 2 - MariaDB",
"database": "mariadb",
"type": "mysql",
"username": "mariadb",
"password": "changeme",
"port": 3306,
"host": "db-mariadb",
"generated_id": "16678124-ff7e-4c97-8c83-0adeff214681"
},
{
"name": "Test database 4 - MongoDB Auth",
"database": "testdbauth",
"type": "mongodb",
"username": "root",
"password": "rootpassword",
"port": 27017,
"host": "db-mongodb-auth",
"generated_id": "16678145-ff7e-4c97-8c83-0adeff214681"
},
{
"name": "Test database 5 - MongoDB",
"database": "testdb",
"type": "mongodb",
"username": "",
"password": "",
"port": 27017,
"host": "db-mongodb",
"generated_id": "16678147-ff7e-4c97-8c83-0adeff214681"
}
]
}
+24 -14
View File
@@ -1,29 +1,39 @@
[[databases]]
name = "Test database 1 - rust"
name = "Test database 1 - PostgreSQL"
database = "devdb"
type = "postgresql"
username = "devuser"
password = "changeme"
port = 5432
host = "db"
host = "db-postgres"
generated_id = "16678159-ff7e-4c97-8c83-0adeff214689"
[[databases]]
name = "Test database 2 - rust"
database = "devdb"
type = "postgresql"
username = "devuser"
password = "changeme"
port = 5432
host = "db"
generated_id = "16678156-ff7e-4c97-8c83-0adeff214654"
[[databases]]
name = "Test database 3 - rust"
name = "Test database 2 - MariaDB"
database = "mariadb"
type = "mysql"
username = "mariadb"
password = "changeme"
port = 3306
host = "db1"
host = "db-mariadb"
generated_id = "16678124-ff7e-4c97-8c83-0adeff214667"
[[databases]]
name = "Test database 3 - MongoDB Auth"
database = "testdbauth"
type = "mongodb"
username = "root"
password = "rootpassword"
port = 27017
host = "db-mongodb-auth"
generated_id = "16678145-ff7e-4c97-8c83-0adeff214681"
[[databases]]
name = "Test database 3 - MongoDB"
database = "testdb"
type = "mongodb"
username = ""
password = ""
port = 27017
host = "db-mongodb"
generated_id = "16678147-ff7e-4c97-8c83-0adeff214681"
+10 -39
View File
@@ -1,53 +1,24 @@
services:
rust-app:
build:
context: .
dockerfile: docker/Dockerfile
target: prod
# build:
# context: .
# dockerfile: docker/Dockerfile
# target: prod
image: portabase/agent:latest
container_name: rust-prod
volumes:
- ./databases.toml:/config/config.toml
- ./databases.json:/config/config.json
environment:
APP_ENV: production
RUST_LOG: info
# APP_ENV: production
LOG: info
TZ: "Europe/Paris"
DATABASES_CONFIG_FILE: "config.toml"
EDGE_KEY: "eyJzZXJ2ZXJVcmwiOiJodHRwOi8vbG9jYWxob3N0Ojg4ODciLCJhZ2VudElkIjoiNzIxZTE0YTQtZDFmZi00OTM4LWFjYzgtNGY1ZWRmMjllODkzIiwicHVibGljS2V5IjoiLS0tLS1CRUdJTiBSU0EgUFVCTElDIEtFWS0tLS0tXG5NSUlCQ2dLQ0FRRUFuYUVKdnVBSExlNGF5d0RmNGplelVobm45VXRkdksyZ3pEMEg2cERJYXczYkJKRkpwVnVDXG5uVFV3MXA3Q2RnOXBzdjZhRnpyOXZPd0J2MjMzckxpdVpCT2lCb2p2Q0QrSlZid3hyTzBRRW5hN2dmaHV1ZGYwXG5VVlJOMkxmK1g1aTkvZzJTNm5xcExoTm1DaGFJNk8ybktYZUNlRmtubEErRUJrNnFoV1FCVGozb05TYTFTOFY1XG40UFRTT2I4NUo3a2k5YllEbXRiNWxrU3dCNXdXOTdtQjg0ZzI2WHAvU3FFcmhKc0NGK3YrN09vTWYzTzJqTTNoXG5XMUQ0MzBPRitWaklwUGdoV09rZy96NXZQUWFHRzhqQ0h4VDlJR0Q0bjhyS05LQ3FTOGNyN2diTGU0cWpNdmhvXG5BQVVvaHpHR2FRNkhlWlJ4S0UvM3J1a2JldnY5dnJ2TTNRSURBUUFCXG4tLS0tLUVORCBSU0EgUFVCTElDIEtFWS0tLS0tXG4ifQ"
# DATABASES_CONFIG_FILE: "config.toml"
EDGE_KEY: "eyJzZXJ2ZXJVcmwiOiJodHRwOi8vbG9jYWxob3N0Ojg4ODciLCJhZ2VudElkIjoiY2VlZmNmNDQtOGE0YS00NjZlLTkwNDEtN2QzNDMzZjRjOTJkIiwicHVibGljS2V5IjoiLS0tLS1CRUdJTiBSU0EgUFVCTElDIEtFWS0tLS0tXG5NSUlCQ2dLQ0FRRUFuYUVKdnVBSExlNGF5d0RmNGplelVobm45VXRkdksyZ3pEMEg2cERJYXczYkJKRkpwVnVDXG5uVFV3MXA3Q2RnOXBzdjZhRnpyOXZPd0J2MjMzckxpdVpCT2lCb2p2Q0QrSlZid3hyTzBRRW5hN2dmaHV1ZGYwXG5VVlJOMkxmK1g1aTkvZzJTNm5xcExoTm1DaGFJNk8ybktYZUNlRmtubEErRUJrNnFoV1FCVGozb05TYTFTOFY1XG40UFRTT2I4NUo3a2k5YllEbXRiNWxrU3dCNXdXOTdtQjg0ZzI2WHAvU3FFcmhKc0NGK3YrN09vTWYzTzJqTTNoXG5XMUQ0MzBPRitWaklwUGdoV09rZy96NXZQUWFHRzhqQ0h4VDlJR0Q0bjhyS05LQ3FTOGNyN2diTGU0cWpNdmhvXG5BQVVvaHpHR2FRNkhlWlJ4S0UvM3J1a2JldnY5dnJ2TTNRSURBUUFCXG4tLS0tLUVORCBSU0EgUFVCTElDIEtFWS0tLS0tXG4ifQ"
extra_hosts:
- "localhost:host-gateway"
networks:
- portabase
db:
image: postgres:17-alpine
ports:
- "5413:5432"
volumes:
- postgres-data:/var/lib/postgresql/data
environment:
- POSTGRES_DB=devdb
- POSTGRES_USER=devuser
- POSTGRES_PASSWORD=changeme
networks:
- portabase
db2:
image: mariadb:latest
ports:
- "3313:3306"
environment:
- MYSQL_DATABASE=mariadb
- MYSQL_USER=mariadb
- MYSQL_PASSWORD=changeme
- MYSQL_RANDOM_ROOT_PASSWORD=yes
volumes:
- mariadb-data:/var/lib/mysql
networks:
- portabase
volumes:
postgres-data:
mariadb-data:
networks:
+75 -26
View File
@@ -4,59 +4,108 @@ services:
context: .
dockerfile: docker/Dockerfile
target: dev
platform: linux/arm64
container_name: rust-dev
volumes:
- .:/app
- ./databases.toml:/config/config.toml
- ./databases.json:/config/config.json
# - ./databases.toml:/config/config.toml
- cargo-registry:/usr/local/cargo/registry
- cargo-git:/usr/local/cargo/git
- cargo-target:/app/target
environment:
APP_ENV: development
RUST_LOG: debug
LOG: debug
TZ: "Europe/Paris"
DATABASES_CONFIG_FILE: "config.toml"
EDGE_KEY: "eyJzZXJ2ZXJVcmwiOiJodHRwOi8vbG9jYWxob3N0Ojg4ODciLCJhZ2VudElkIjoiOThlZmY1YjEtMDQyZC00MDkxLWJkYjctZTgzYmM0YjczNjAxIiwicHVibGljS2V5IjoiLS0tLS1CRUdJTiBSU0EgUFVCTElDIEtFWS0tLS0tXG5NSUlCQ2dLQ0FRRUFuYUVKdnVBSExlNGF5d0RmNGplelVobm45VXRkdksyZ3pEMEg2cERJYXczYkJKRkpwVnVDXG5uVFV3MXA3Q2RnOXBzdjZhRnpyOXZPd0J2MjMzckxpdVpCT2lCb2p2Q0QrSlZid3hyTzBRRW5hN2dmaHV1ZGYwXG5VVlJOMkxmK1g1aTkvZzJTNm5xcExoTm1DaGFJNk8ybktYZUNlRmtubEErRUJrNnFoV1FCVGozb05TYTFTOFY1XG40UFRTT2I4NUo3a2k5YllEbXRiNWxrU3dCNXdXOTdtQjg0ZzI2WHAvU3FFcmhKc0NGK3YrN09vTWYzTzJqTTNoXG5XMUQ0MzBPRitWaklwUGdoV09rZy96NXZQUWFHRzhqQ0h4VDlJR0Q0bjhyS05LQ3FTOGNyN2diTGU0cWpNdmhvXG5BQVVvaHpHR2FRNkhlWlJ4S0UvM3J1a2JldnY5dnJ2TTNRSURBUUFCXG4tLS0tLUVORCBSU0EgUFVCTElDIEtFWS0tLS0tXG4ifQ"
EDGE_KEY: "eyJzZXJ2ZXJVcmwiOiJodHRwOi8vbG9jYWxob3N0Ojg4ODciLCJhZ2VudElkIjoiNmExMzc0M2ItYjMwZS00Zjg4LWJjY2EtMmMwZjE2NWVjYjQxIiwicHVibGljS2V5IjoiLS0tLS1CRUdJTiBSU0EgUFVCTElDIEtFWS0tLS0tXG5NSUlCQ2dLQ0FRRUE5TWV4M2pmdnVLdFB5YU1ERnh2Ulp2dmd3YkRJQ2JzQi81Wll5NDNSVVRBaXZRYjJiSDdYXG5qRHBQd1lJeCs4UFBrbHlRbDVMQzV1UWZEaCs4SVd4OG1LZ3FvMXpWMkdiZXdGbEdEWFYxVEdyU1ZEU25aSWR4XG52bWdYc29EeXhVMlJvWUFUMS9YMWxuc2YxenZKdkFMTkhXdEhRdk42SjVDZTFSMmFsendVRGFEVXlJNzRmSldQXG5tNTh0SDMrYklXL0VVTXdjaWNxM0oySWw3Vm9KNkZNUHJQL1ZSOWEvdFF1SU1qa200MXpFY2NscExPa2luRkxuXG54NmVUWkFSZUpya2UrbnRvZ2t4TGEyRWV5a1lUNzB4V3hKNWp5ZExBVnRvNkkyQlVLVVJoTkowTUFaU29NYUtvXG5iMGJRcnY1UzExZWllMnMrT2I3aTYzSFpkVUx0UmV1MVJ3SURBUUFCXG4tLS0tLUVORCBSU0EgUFVCTElDIEtFWS0tLS0tXG4ifQ"
#POOLING: 1
#DATABASES_CONFIG_FILE: "config.toml"
extra_hosts:
- "localhost:host-gateway"
networks:
- portabase
db:
image: postgres:17-alpine
# db-postgres:
# container_name: db-postgres
# image: postgres:17-alpine
# ports:
# - "5436:5432"
# volumes:
# - postgres-data:/var/lib/postgresql/data
# environment:
# - POSTGRES_DB=devdb
# - POSTGRES_USER=devuser
# - POSTGRES_PASSWORD=changeme
# networks:
# - portabase
#
# db-mariadb:
# container_name: db-mariadb
# image: mariadb:latest
# ports:
# - "3311:3306"
# environment:
# - MYSQL_DATABASE=mariadb
# - MYSQL_USER=mariadb
# - MYSQL_PASSWORD=changeme
# - MYSQL_RANDOM_ROOT_PASSWORD=yes
# volumes:
# - mariadb-data:/var/lib/mysql
# networks:
# - portabase
db-mongodb-auth:
container_name: db-mongodb-auth
image: mongo:latest
ports:
- "5436:5432"
volumes:
- postgres-data:/var/lib/postgresql/data
- "27082:27017"
environment:
- POSTGRES_DB=devdb
- POSTGRES_USER=devuser
- POSTGRES_PASSWORD=changeme
MONGO_INITDB_ROOT_USERNAME: root
MONGO_INITDB_ROOT_PASSWORD: rootpassword
MONGO_INITDB_DATABASE: testdbauth
command: mongod --auth
networks:
- portabase
db2:
image: mariadb:latest
ports:
- "3311:3306"
environment:
- MYSQL_DATABASE=mariadb
- MYSQL_USER=mariadb
- MYSQL_PASSWORD=changeme
- MYSQL_RANDOM_ROOT_PASSWORD=yes
volumes:
- mariadb-data:/var/lib/mysql
- mongodb-data-auth:/data/db
healthcheck:
test: [ "CMD", "mongo", "--eval", "db.adminCommand('ping')" ]
interval: 5s
timeout: 5s
retries: 10
db-mongodb:
container_name: db-mongodb
image: mongo:latest
ports:
- "27083:27017"
volumes:
- mongodb-data:/data/db
healthcheck:
test: [ "CMD", "mongosh", "--eval", "db.adminCommand('ping')" ]
interval: 5s
timeout: 5s
retries: 10
environment:
MONGO_INITDB_DATABASE: testdb
networks:
- portabase
volumes:
cargo-registry:
cargo-git:
cargo-target:
postgres-data:
mariadb-data:
# postgres-data:
# mariadb-data:
mongodb-data:
mongodb-data-auth:
networks:
portabase:
name: portabase_network
external: true
external: true
# docker network create portabase_network
+34 -108
View File
@@ -1,115 +1,8 @@
## =========================
## Base image (shared)
## =========================
#FROM rust:1.92.0 AS base
#
#RUN apt-get update && DEBIAN_FRONTEND=noninteractive apt-get install -y \
# pkg-config \
# libssl-dev \
# tzdata \
# redis-server \
# ca-certificates \
# curl \
# && apt-get clean \
# && rm -rf /var/lib/apt/lists/*
#
#
#
## ========= Install PostgreSQL client binaries (versions 12-18) =========
## Pre-downloaded binaries from assets/tools/ - no network download needed
#ARG TARGETARCH
#RUN mkdir -p /usr/lib/postgresql/12/bin /usr/lib/postgresql/13/bin \
# /usr/lib/postgresql/14/bin /usr/lib/postgresql/15/bin \
# /usr/lib/postgresql/16/bin /usr/lib/postgresql/17/bin \
# /usr/lib/postgresql/18/bin
#
## Copy pre-downloaded PostgreSQL binaries based on architecture
#COPY assets/tools/x64/postgresql/ /tmp/pg-x64/
#COPY assets/tools/arm/postgresql/ /tmp/pg-arm/
#RUN if [ "$TARGETARCH" = "amd64" ]; then \
# cp -r /tmp/pg-x64/postgresql-12/bin/* /usr/lib/postgresql/12/bin/ && \
# cp -r /tmp/pg-x64/postgresql-13/bin/* /usr/lib/postgresql/13/bin/ && \
# cp -r /tmp/pg-x64/postgresql-14/bin/* /usr/lib/postgresql/14/bin/ && \
# cp -r /tmp/pg-x64/postgresql-15/bin/* /usr/lib/postgresql/15/bin/ && \
# cp -r /tmp/pg-x64/postgresql-16/bin/* /usr/lib/postgresql/16/bin/ && \
# cp -r /tmp/pg-x64/postgresql-17/bin/* /usr/lib/postgresql/17/bin/ && \
# cp -r /tmp/pg-x64/postgresql-18/bin/* /usr/lib/postgresql/18/bin/; \
# elif [ "$TARGETARCH" = "arm64" ]; then \
# cp -r /tmp/pg-arm/postgresql-12/bin/* /usr/lib/postgresql/12/bin/ && \
# cp -r /tmp/pg-arm/postgresql-13/bin/* /usr/lib/postgresql/13/bin/ && \
# cp -r /tmp/pg-arm/postgresql-14/bin/* /usr/lib/postgresql/14/bin/ && \
# cp -r /tmp/pg-arm/postgresql-15/bin/* /usr/lib/postgresql/15/bin/ && \
# cp -r /tmp/pg-arm/postgresql-16/bin/* /usr/lib/postgresql/16/bin/ && \
# cp -r /tmp/pg-arm/postgresql-17/bin/* /usr/lib/postgresql/17/bin/ && \
# cp -r /tmp/pg-arm/postgresql-18/bin/* /usr/lib/postgresql/18/bin/; \
# fi && \
# rm -rf /tmp/pg-x64 /tmp/pg-arm && \
# chmod +x /usr/lib/postgresql/*/bin/*
#
#
#
#WORKDIR /app
#
## =========================
## Development image
## =========================
#FROM base AS dev
#
#RUN cargo install cargo-watch
#
## Pre-cache dependencies
#COPY Cargo.toml Cargo.lock ./
#RUN mkdir src && echo "fn main() {}" > src/main.rs
#RUN cargo build
#RUN rm -rf src
#
#COPY entrypoint.sh /entrypoint.sh
#RUN chmod +x /entrypoint.sh
#
#CMD ["/entrypoint.sh"]
#
## =========================
## Builder (production)
## =========================
#FROM base AS builder
#
#COPY . .
#RUN cargo build --release
#
#RUN echo "APP_VERSION=$(cargo pkgid | awk -F# '{print $2}')" > /app/version.env
#
## =========================
## Runtime (production)
## =========================
#FROM debian:bookworm-slim AS prod
##FROM ubuntu:24.04 AS prod
#
#
#RUN apt-get update && apt-get install -y \
# redis-server \
# ca-certificates \
# tzdata \
# postgresql-client \
# libpq5 \
# mariadb-client \
# && rm -rf /var/lib/apt/lists/*
#
#WORKDIR /app
#
#COPY --from=builder /app/target/release/app /usr/local/bin/app
#COPY --from=builder /app/version.env /app/version.env
#COPY entrypoint.sh /entrypoint.sh
#RUN chmod +x /entrypoint.sh
#
#ENV APP_ENV=production
#
#CMD ["/entrypoint.sh"]
# =========================
# Base image (shared)
# =========================
FROM rust:1.92.0 AS base
# Install common dependencies
RUN apt-get update && DEBIAN_FRONTEND=noninteractive apt-get install -y \
pkg-config \
libssl-dev \
@@ -117,15 +10,20 @@ RUN apt-get update && DEBIAN_FRONTEND=noninteractive apt-get install -y \
redis-server \
ca-certificates \
libpq5 \
libreadline8 \
libncurses6 \
zlib1g \
curl \
mariadb-client \
&& apt-get clean \
&& rm -rf /var/lib/apt/lists/*
ARG TARGETARCH
# =========================
# PostgreSQL client binaries (versions 12-18)
# =========================
ARG TARGETARCH
RUN for v in 12 13 14 15 16 17 18; do \
mkdir -p /usr/lib/postgresql/$v/bin; \
@@ -147,6 +45,29 @@ RUN if [ "$TARGETARCH" = "amd64" ]; then \
chmod +x /usr/lib/postgresql/*/bin/*
# =========================
# MongoDB client binaries
# =========================
RUN apt-get update && \
if [ "$TARGETARCH" = "amd64" ]; then \
wget -q https://fastdl.mongodb.org/tools/db/mongodb-database-tools-debian12-x86_64-100.10.0.deb \
-O /tmp/mongodb-database-tools.deb; \
elif [ "$TARGETARCH" = "arm64" ]; then \
wget -q https://fastdl.mongodb.org/tools/db/mongodb-database-tools-ubuntu2204-arm64-100.10.0.deb \
-O /tmp/mongodb-database-tools.deb; \
fi && \
dpkg -i /tmp/mongodb-database-tools.deb || \
apt-get install -f -y --no-install-recommends && \
rm -f /tmp/mongodb-database-tools.deb && \
rm -rf /var/lib/apt/lists/* && \
mkdir -p /usr/local/mongodb/bin && \
if [ -f /usr/bin/mongodump ]; then \
ln -sf /usr/bin/mongodump /usr/local/mongodb/bin/mongodump; \
fi && \
if [ -f /usr/bin/mongorestore ]; then \
ln -sf /usr/bin/mongorestore /usr/local/mongodb/bin/mongorestore; \
fi
WORKDIR /app
# =========================
@@ -187,9 +108,13 @@ RUN apt-get update && apt-get install -y \
ca-certificates \
tzdata \
libpq5 \
libreadline8 \
libncurses6 \
zlib1g \
mariadb-client \
&& rm -rf /var/lib/apt/lists/*
WORKDIR /app
COPY --from=builder /app/target/release/app /usr/local/bin/app
@@ -198,6 +123,7 @@ COPY entrypoint.sh /entrypoint.sh
RUN chmod +x /entrypoint.sh
COPY --from=base /usr/lib/postgresql/ /usr/lib/postgresql/
COPY --from=base /usr/local/mongodb/bin /usr/local/mongodb/bin
ENV APP_ENV=production
+12
View File
@@ -0,0 +1,12 @@
# Seed instructions
## MongoDB
```bash
make seed-mongo
make seed-mongo-auth
make seed-mysql
make seed-postgres
make seed-postgres-1gb
make seed-all
```
+71
View File
@@ -0,0 +1,71 @@
#!/usr/bin/env bash
# Usage:
# ./seed-mongo.sh -> no auth
# ./seed-mongo.sh auth -> with auth
MODE=${1:-noauth} # default: noauth
# Default values (can be overridden via .env)
MONGO_DB=${MONGO_DB:-testdb}
MONGO_PORT=${MONGO_PORT:-27017}
# Auth variables
MONGO_AUTH_USERNAME=${MONGO_AUTH_USERNAME:-root}
MONGO_AUTH_PASSWORD=${MONGO_AUTH_PASSWORD:-rootpassword}
MONGO_AUTH_DB=${MONGO_AUTH_DB:-testdbauth}
NUM_USERS=${NUM_USERS:-5000}
# Build connection string
if [ "$MODE" = "auth" ]; then
MONGO_CONTAINER=${MONGO_AUTH_CONTAINER:-db-mongodb-auth}
URI="mongodb://$MONGO_AUTH_USERNAME:$MONGO_AUTH_PASSWORD@localhost:$MONGO_PORT/$MONGO_AUTH_DB?authSource=admin"
echo "Using authenticated MongoDB..."
else
MONGO_CONTAINER=${MONGO_CONTAINER:-db-mongodb}
URI="mongodb://localhost:$MONGO_PORT/$MONGO_DB"
echo "Using non-authenticated MongoDB..."
fi
# Create temp JS seed file
SEED_FILE=$(mktemp /tmp/seed.XXXX.js)
cat <<EOF > "$SEED_FILE"
db = connect("$URI");
// Drop collections if exist
if (db.users) db.users.drop();
if (db.products) db.products.drop();
// Generate users
const users = [];
for (let i = 1; i <= $NUM_USERS; i++) {
users.push({
username: "user" + i,
email: "user" + i + "@example.com",
age: Math.floor(Math.random() * 60) + 18,
active: Math.random() > 0.5,
createdAt: new Date()
});
}
db.users.insertMany(users);
// Generate products
const products = [];
for (let i = 1; i <= Math.floor($NUM_USERS/10); i++) {
products.push({
name: "Product " + i,
price: parseFloat((Math.random()*100).toFixed(2)),
stock: Math.floor(Math.random()*500),
createdAt: new Date()
});
}
db.products.insertMany(products);
print("Seed complete: " + users.length + " users, " + products.length + " products.");
EOF
docker cp "$SEED_FILE" "$MONGO_CONTAINER:/seed.js"
docker exec -it "$MONGO_CONTAINER" mongosh /seed.js
rm "$SEED_FILE"
+17 -16
View File
@@ -1,16 +1,15 @@
#![allow(dead_code)]
use crate::domain::postgres::PostgresDatabase;
use crate::services::config::DatabaseConfig;
use crate::domain::mongodb::database::MongoDatabase;
use crate::domain::mysql::database::MySQLDatabase;
use crate::domain::postgres::database::PostgresDatabase;
use crate::domain::postgres::{detect_format_from_file, detect_format_from_size};
use crate::services::config::{DatabaseConfig, DbType};
use anyhow::Result;
use std::path::{Path, PathBuf};
use std::sync::Arc;
use crate::domain::mysql::MySQLDatabase;
#[async_trait::async_trait]
pub trait Database: Send + Sync {
fn file_extension(&self) -> &'static str;
async fn ping(&self) -> Result<bool>;
async fn backup(&self, backup_dir: &Path) -> Result<PathBuf>;
async fn restore(&self, restore_file: &Path) -> Result<()>;
@@ -20,24 +19,26 @@ pub struct DatabaseFactory;
impl DatabaseFactory {
pub async fn create_for_backup(cfg: DatabaseConfig) -> Arc<dyn Database> {
match cfg.db_type.as_str() {
"postgresql" => {
let format = PostgresDatabase::detect_format_from_size(&cfg).await;
match cfg.db_type {
DbType::Postgresql => {
let format = detect_format_from_size(&cfg).await;
Arc::new(PostgresDatabase::new(cfg, format))
}
"mysql" => Arc::new(MySQLDatabase::new(cfg)),
_ => panic!("Unsupported DB type: {}", cfg.db_type),
DbType::Mysql => Arc::new(MySQLDatabase::new(cfg)),
DbType::Mariadb => Arc::new(MySQLDatabase::new(cfg)),
DbType::MongoDB => Arc::new(MongoDatabase::new(cfg)),
}
}
pub async fn create_for_restore(cfg: DatabaseConfig, restore_file: &Path) -> Arc<dyn Database> {
match cfg.db_type.as_str() {
"postgresql" => {
let format = PostgresDatabase::detect_format_from_file(restore_file);
match cfg.db_type {
DbType::Postgresql => {
let format = detect_format_from_file(restore_file);
Arc::new(PostgresDatabase::new(cfg, format))
}
"mysql" => Arc::new(MySQLDatabase::new(cfg)),
_ => panic!("Unsupported DB type: {}", cfg.db_type),
DbType::Mysql => Arc::new(MySQLDatabase::new(cfg)),
DbType::Mariadb => Arc::new(MySQLDatabase::new(cfg)),
DbType::MongoDB => Arc::new(MongoDatabase::new(cfg)),
}
}
}
+2
View File
@@ -1,3 +1,5 @@
pub mod factory;
pub mod postgres;
pub mod mysql;
mod mongodb;
+36
View File
@@ -0,0 +1,36 @@
use crate::domain::mongodb::connection::{get_mongo_uri, select_mongo_path};
use crate::services::config::DatabaseConfig;
use anyhow::{Context, Result};
use std::path::PathBuf;
use std::process::Command;
use tracing::{debug, error, info};
pub async fn run(
cfg: DatabaseConfig,
backup_dir: PathBuf,
file_extension: &'static str,
) -> Result<PathBuf> {
tokio::task::spawn_blocking(move || -> Result<PathBuf> {
debug!("Starting MongoDB backup for database {}", cfg.name);
let file_path = backup_dir.join(format!("{}{}", cfg.generated_id, file_extension));
let mongodump = select_mongo_path().join("mongodump");
let uri = get_mongo_uri(cfg.clone());
let output = Command::new(mongodump)
.arg(format!("--uri={}", uri))
.arg(format!("--archive={}", file_path.display()))
.arg("--gzip")
.output()
.context("MongoDB backup failed")?;
if !output.status.success() {
let stderr = String::from_utf8_lossy(&output.stderr);
error!("MongoDB backup failed for {}: {}", cfg.name, stderr);
anyhow::bail!("MongoDB backup failed for {}: {}", cfg.name, stderr);
}
info!("MongoDB backup completed for {}", cfg.name);
Ok(file_path)
})
.await?
}
+28
View File
@@ -0,0 +1,28 @@
use crate::services::config::DatabaseConfig;
use anyhow::Result;
use mongodb::Client;
pub async fn connect(cfg: DatabaseConfig) -> Result<Client> {
let uri = get_mongo_uri(cfg);
let mut options = mongodb::options::ClientOptions::parse(&uri).await?;
options.server_selection_timeout = Some(std::time::Duration::from_secs(3));
options.connect_timeout = Some(std::time::Duration::from_secs(3));
let client = Client::with_options(options)?;
Ok(client)
}
pub fn select_mongo_path() -> std::path::PathBuf {
"/usr/local/mongodb/bin".to_string().into()
}
pub fn get_mongo_uri(cfg: DatabaseConfig) -> String {
if cfg.username.is_empty() {
format!("mongodb://{}:{}/{}", cfg.host, cfg.port, cfg.database)
} else {
format!(
"mongodb://{}:{}@{}:{}/{}?authSource=admin",
cfg.username, cfg.password, cfg.host, cfg.port, cfg.database
)
}
}
+48
View File
@@ -0,0 +1,48 @@
use anyhow::Result;
use async_trait::async_trait;
use std::path::{Path, PathBuf};
use super::{backup, ping, restore};
use crate::domain::factory::Database;
use crate::services::config::DatabaseConfig;
use crate::utils::locks::{DbOpLock, FileLock};
pub struct MongoDatabase {
cfg: DatabaseConfig,
}
impl MongoDatabase {
pub fn new(cfg: DatabaseConfig) -> Self {
Self { cfg }
}
}
#[async_trait]
impl Database for MongoDatabase {
fn file_extension(&self) -> &'static str {
".archive.gz"
}
async fn ping(&self) -> Result<bool> {
ping::run(self.cfg.clone()).await
}
async fn backup(&self, dir: &Path) -> Result<PathBuf> {
FileLock::acquire(&self.cfg.generated_id, DbOpLock::Backup.as_str()).await?;
let res = backup::run(
self.cfg.clone(),
dir.to_path_buf(),
self.file_extension(),
)
.await;
FileLock::release(&self.cfg.generated_id).await?;
res
}
async fn restore(&self, file: &Path) -> Result<()> {
FileLock::acquire(&self.cfg.generated_id, DbOpLock::Restore.as_str()).await?;
let res = restore::run(self.cfg.clone(), file.to_path_buf()).await;
FileLock::release(&self.cfg.generated_id).await?;
res
}
}
+5
View File
@@ -0,0 +1,5 @@
mod backup;
mod restore;
pub mod database;
mod ping;
mod connection;
+24
View File
@@ -0,0 +1,24 @@
#![allow(dead_code)]
use crate::services::config::DatabaseConfig;
use anyhow::Result;
use mongodb::bson::doc;
use tracing::{error};
use crate::domain::mongodb::connection::connect;
pub async fn run(cfg: DatabaseConfig) -> Result<bool> {
let client = connect(cfg.clone()).await?;
let db_name = if cfg.username.is_empty() { &cfg.database } else { "admin" };
match client.database(db_name).run_command(doc! {"ping": 1}).await {
Ok(_) => Ok(true),
Err(e) => {
error!("--- MongoDB Connection Error Details ---");
error!("Target Host: {}:{}", cfg.host, cfg.port);
error!("Error Kind: {:?}", e.kind);
error!("Full Error: {}", e);
error!("Check you database network connectivity");
error!("----------------------------------------");
Err(anyhow::anyhow!("Ping failed for {}: {}", cfg.name, e))
}
}
}
+33
View File
@@ -0,0 +1,33 @@
use crate::domain::mongodb::connection::{get_mongo_uri, select_mongo_path};
use crate::services::config::DatabaseConfig;
use anyhow::{Context, Result};
use std::path::PathBuf;
use std::process::Command;
use tracing::{debug, error, info};
pub async fn run(cfg: DatabaseConfig, restore_file: PathBuf) -> Result<()> {
tokio::task::spawn_blocking(move || -> Result<()> {
debug!("Starting MongoDB restore for database {}", cfg.name);
let mongorestore = select_mongo_path().join("mongorestore");
let uri = get_mongo_uri(cfg.clone());
let output = Command::new(mongorestore)
.arg(format!("--uri={}", uri))
.arg(format!("--archive={}", restore_file.display()))
.arg("--gzip")
.arg("--drop")
.output()
.with_context(|| format!("Failed to run mongorestore for {}", cfg.name))?;
if !output.status.success() {
let stderr = String::from_utf8_lossy(&output.stderr);
error!("MongoDB restore failed for {}: {}", cfg.name, stderr);
anyhow::bail!("MongoDB restore failed for : {}", cfg.name);
}
info!("MongoDB restore completed for {}", cfg.name);
Ok(())
})
.await?
}
-140
View File
@@ -1,140 +0,0 @@
use crate::domain::factory::Database;
use crate::services::config::DatabaseConfig;
use anyhow::{Context, Result};
use std::collections::HashMap;
use std::io::Write;
use std::path::{Path, PathBuf};
use std::process::Command;
pub struct MySQLDatabase {
cfg: DatabaseConfig,
}
impl MySQLDatabase {
pub fn new(cfg: DatabaseConfig) -> Self {
Self { cfg }
}
fn build_env(&self) -> HashMap<String, String> {
let mut envs = std::env::vars().collect::<HashMap<_, _>>();
envs.insert("MYSQL_PWD".to_string(), self.cfg.password.clone());
envs
}
}
#[async_trait::async_trait]
impl Database for MySQLDatabase {
fn file_extension(&self) -> &'static str {
".sql"
}
async fn ping(&self) -> Result<bool> {
let output = Command::new("mysqladmin")
.arg("--host")
.arg(&self.cfg.host)
.arg("--port")
.arg(self.cfg.port.to_string())
.arg("--user")
.arg(&self.cfg.username)
.arg("ping")
.envs(self.build_env())
.output()
.with_context(|| format!("Failed to ping MySQL server {}", self.cfg.name))?;
Ok(output.status.success())
}
async fn backup(&self, backup_dir: &Path) -> Result<PathBuf> {
let file_path = backup_dir.join(format!(
"{}{}",
self.cfg.generated_id,
self.file_extension()
));
let output = Command::new("mysqldump")
.arg("--host")
.arg(&self.cfg.host)
.arg("--port")
.arg(self.cfg.port.to_string())
.arg("--user")
.arg(&self.cfg.username)
.arg("--routines")
.arg("--events")
.arg("--triggers")
.arg("--verbose")
.arg("--single-transaction")
.arg("--quick")
.arg("--add-drop-database")
.arg("--databases")
.arg(&self.cfg.database)
.arg("-r")
.arg(&file_path)
.envs(self.build_env())
.output()
.with_context(|| format!("Failed to run mysqldump for {}", self.cfg.name))?;
if !output.status.success() {
let stderr = String::from_utf8_lossy(&output.stderr);
anyhow::bail!("MySQL backup failed for {}: {}", self.cfg.name, stderr);
}
Ok(file_path)
}
async fn restore(&self, restore_file: &Path) -> Result<()> {
let sql_content = tokio::fs::read_to_string(restore_file)
.await
.with_context(|| format!("Failed to read restore file {}", restore_file.display()))?;
let drop_create_cmd = format!(
"DROP DATABASE IF EXISTS {0}; CREATE DATABASE {0};",
self.cfg.database
);
let drop_status = Command::new("mysql")
.arg("--host")
.arg(&self.cfg.host)
.arg("--port")
.arg(self.cfg.port.to_string())
.arg("--user")
.arg(&self.cfg.username)
.arg("-e")
.arg(&drop_create_cmd)
.env("MYSQL_PWD", &self.cfg.password)
.status()
.with_context(|| format!("Failed to drop/recreate database {}", self.cfg.name))?;
if !drop_status.success() {
anyhow::bail!("Failed to drop/recreate database {}", self.cfg.name);
}
let mut child = Command::new("mysql")
.arg("--host")
.arg(&self.cfg.host)
.arg("--port")
.arg(self.cfg.port.to_string())
.arg("--user")
.arg(&self.cfg.username)
.arg(&self.cfg.database)
.env("MYSQL_PWD", &self.cfg.password)
.stdin(std::process::Stdio::piped())
.spawn()
.with_context(|| format!("Failed to start mysql restore for {}", self.cfg.name))?;
let mut stdin = child.stdin.take().context("Failed to open child stdin")?;
stdin.write_all(sql_content.as_bytes())?;
stdin.flush()?;
drop(stdin);
let output = child
.wait_with_output()
.with_context(|| format!("Failed to complete mysql restore for {}", self.cfg.name))?;
if !output.status.success() {
let stderr = String::from_utf8_lossy(&output.stderr);
anyhow::bail!("MySQL restore failed for {}: {}", self.cfg.name, stderr);
}
Ok(())
}
}
+63
View File
@@ -0,0 +1,63 @@
use crate::domain::mysql::connection::server_version;
use crate::services::config::DatabaseConfig;
use anyhow::{Context, Result};
use std::collections::HashMap;
use std::path::PathBuf;
use std::process::Command;
use tracing::{debug, error, info};
pub async fn run(
cfg: DatabaseConfig,
backup_dir: PathBuf,
env: HashMap<String, String>,
file_extension: &'static str,
) -> Result<PathBuf> {
tokio::task::spawn_blocking(move || -> Result<PathBuf> {
debug!("Starting backup for database {}", cfg.name);
let version = match futures::executor::block_on(server_version(&cfg)) {
Ok(v) => {
debug!("Mysql version detected: {}", v);
v
}
Err(e) => {
error!("Failed to get server version for {}: {:?}", cfg.name, e);
return Err(e.into());
}
};
info!("Mysql version found: {}", version);
let file_path = backup_dir.join(format!("{}{}", cfg.generated_id, file_extension));
let output = Command::new("mysqldump")
.arg("--host")
.arg(cfg.host)
.arg("--port")
.arg(cfg.port.to_string())
.arg("--user")
.arg(cfg.username)
.arg("--routines")
.arg("--events")
.arg("--triggers")
.arg("--verbose")
.arg("--single-transaction")
.arg("--quick")
.arg("--add-drop-database")
.arg("--databases")
.arg(cfg.database)
.arg("-r")
.arg(&file_path)
.envs(env)
.output()
.with_context(|| format!("Failed to run mysqldump for {}", cfg.name))?;
if !output.status.success() {
let stderr = String::from_utf8_lossy(&output.stderr);
anyhow::bail!("MySQL backup failed for {}: {}", cfg.name, stderr);
}
Ok(file_path)
})
.await?
}
+27
View File
@@ -0,0 +1,27 @@
use crate::services::config::DatabaseConfig;
use std::process::Command;
use anyhow::Result;
pub async fn server_version(cfg: &DatabaseConfig) -> Result<String> {
let output = Command::new("mysql")
.arg("--host").arg(&cfg.host)
.arg("--port").arg(cfg.port.to_string())
.arg("--user").arg(&cfg.username)
.arg("-e").arg("SELECT VERSION();")
.env("MYSQL_PWD", &cfg.password)
.output()?;
if !output.status.success() {
let stderr = String::from_utf8_lossy(&output.stderr);
anyhow::bail!("Version query failed: {}", stderr);
}
let version = String::from_utf8_lossy(&output.stdout)
.lines()
.nth(1) // skip column header
.unwrap_or_default()
.trim()
.to_string();
Ok(version)
}
+52
View File
@@ -0,0 +1,52 @@
use std::collections::HashMap;
use anyhow::Result;
use async_trait::async_trait;
use std::path::{Path, PathBuf};
use super::{
backup,
ping, restore,
};
use crate::domain::factory::Database;
use crate::services::config::DatabaseConfig;
use crate::utils::locks::{DbOpLock, FileLock};
pub struct MySQLDatabase {
cfg: DatabaseConfig,
}
impl MySQLDatabase {
pub fn new(cfg: DatabaseConfig) -> Self {
Self { cfg }
}
fn build_env(&self) -> HashMap<String, String> {
let mut envs = std::env::vars().collect::<HashMap<_, _>>();
envs.insert("MYSQL_PWD".to_string(), self.cfg.password.clone());
envs
}
}
#[async_trait]
impl Database for MySQLDatabase {
fn file_extension(&self) -> &'static str {
".sql"
}
async fn ping(&self) -> Result<bool> {
ping::run(self.cfg.clone(), self.build_env().clone()).await
}
async fn backup(&self, dir: &Path) -> Result<PathBuf> {
FileLock::acquire(&self.cfg.generated_id, DbOpLock::Backup.as_str()).await?;
let res = backup::run(self.cfg.clone(), dir.to_path_buf(), self.build_env().clone(), self.file_extension()).await;
FileLock::release(&self.cfg.generated_id).await?;
res
}
async fn restore(&self, file: &Path) -> Result<()> {
FileLock::acquire(&self.cfg.generated_id, DbOpLock::Restore.as_str()).await?;
let res = restore::run(self.cfg.clone(), file.to_path_buf()).await;
FileLock::release(&self.cfg.generated_id).await?;
res
}
}
+5
View File
@@ -0,0 +1,5 @@
pub mod backup;
pub mod database;
mod restore;
mod ping;
mod connection;
+19
View File
@@ -0,0 +1,19 @@
use crate::services::config::DatabaseConfig;
use anyhow::Context;
use std::collections::HashMap;
use std::process::Command;
pub async fn run(cfg: DatabaseConfig, env: HashMap<String, String>) -> anyhow::Result<bool> {
let output = Command::new("mysqladmin")
.arg("--host")
.arg(cfg.host)
.arg("--port")
.arg(cfg.port.to_string())
.arg("--user")
.arg(cfg.username)
.arg("ping")
.envs(env)
.output()
.with_context(|| format!("Failed to ping MySQL server {}", cfg.name))?;
Ok(output.status.success())
}
+81
View File
@@ -0,0 +1,81 @@
use std::fs::File;
use std::io::{Read, Write};
use anyhow::{Context, Result};
use tracing::{debug, error, info};
use std::path::PathBuf;
use std::process::Command;
use crate::services::config::DatabaseConfig;
pub async fn run(cfg: DatabaseConfig, restore_file: PathBuf) -> Result<()> {
let handle = tokio::task::spawn_blocking(move || -> Result<()> {
debug!("Starting restore for database {}", cfg.name);
let mut sql_content = String::new();
let mut file = File::open(&restore_file)
.with_context(|| format!("Failed to open restore file {}", restore_file.display()))?;
file.read_to_string(&mut sql_content)
.with_context(|| format!("Failed to read restore file {}", restore_file.display()))?;
let drop_create_cmd = format!(
"DROP DATABASE IF EXISTS {0}; CREATE DATABASE {0};",
cfg.database
);
let drop_status = Command::new("mysql")
.arg("--host")
.arg(&cfg.host)
.arg("--port")
.arg(cfg.port.to_string())
.arg("--user")
.arg(&cfg.username)
.arg("-e")
.arg(&drop_create_cmd)
.env("MYSQL_PWD", &cfg.password)
.status()
.with_context(|| format!("Failed to drop/recreate database {}", cfg.name))?;
if !drop_status.success() {
error!("Drop/create database failed for {}", cfg.name);
anyhow::bail!("Failed to drop/recreate database {}", cfg.name);
}
info!("Database {} dropped and recreated", cfg.name);
let mut child = Command::new("mysql")
.arg("--host")
.arg(&cfg.host)
.arg("--port")
.arg(cfg.port.to_string())
.arg("--user")
.arg(&cfg.username)
.arg(&cfg.database)
.env("MYSQL_PWD", &cfg.password)
.stdin(std::process::Stdio::piped())
.spawn()
.with_context(|| format!("Failed to start mysql restore for {}", cfg.name))?;
let mut stdin = child.stdin.take().context("Failed to open child stdin")?;
stdin.write_all(sql_content.as_bytes())
.context("Failed to write SQL content to mysql stdin")?;
stdin.flush()?;
drop(stdin);
let output = child
.wait_with_output()
.with_context(|| format!("Failed to complete mysql restore for {}", cfg.name))?;
if !output.status.success() {
let stderr = String::from_utf8_lossy(&output.stderr);
error!("MySQL restore failed for {}: {}", cfg.name, stderr);
anyhow::bail!("MySQL restore failed for {}", cfg.name);
}
info!("Restore finished successfully for database {}", cfg.name);
Ok(())
});
handle
.await??;
Ok(())
}
-301
View File
@@ -1,301 +0,0 @@
#![allow(dead_code)]
use crate::domain::factory::Database;
use crate::services::config::DatabaseConfig;
use anyhow::{Context as AnyhowContext, Result};
use async_trait::async_trait;
use flate2::Compression;
use flate2::write::GzEncoder;
use log::info;
use std::path::{Path, PathBuf};
use std::process::Command;
use tracing::debug;
#[derive(Clone, Copy)]
pub enum PostgresDumpFormat {
Fc, // legacy
Fd, // directory format
}
pub struct PostgresDatabase {
cfg: DatabaseConfig,
format: PostgresDumpFormat,
}
impl PostgresDatabase {
pub fn new(cfg: DatabaseConfig, format: PostgresDumpFormat) -> Self {
Self { cfg, format }
}
pub fn detect_format_from_file(restore_file: &Path) -> PostgresDumpFormat {
match restore_file.extension().and_then(|e| e.to_str()) {
Some("dump") => PostgresDumpFormat::Fc,
Some("gz") => PostgresDumpFormat::Fd,
// Some("tar.gz") => PostgresDumpFormat::Fd,
_ => PostgresDumpFormat::Fc,
}
}
pub async fn detect_format_from_size(cfg: &DatabaseConfig) -> PostgresDumpFormat {
let url = format!(
"host={} port={} user={} password={} dbname={}",
cfg.host, cfg.port, cfg.username, cfg.password, cfg.database
);
let output = std::process::Command::new("psql")
.arg(&url)
.arg("-t")
.arg("-c")
.arg("SELECT pg_database_size(current_database());")
.output();
match output {
Ok(out) if out.status.success() => {
let size_bytes: i64 = String::from_utf8_lossy(&out.stdout)
.trim()
.parse()
.unwrap_or(0);
// > 1 Go
if size_bytes > 1_000_000_000 {
PostgresDumpFormat::Fd
} else {
PostgresDumpFormat::Fc
}
}
_ => PostgresDumpFormat::Fc, // fallback legacy
}
}
fn get_postgres_server_version(cfg: &DatabaseConfig) -> Result<String, anyhow::Error> {
let output = Command::new("/usr/lib/postgresql/17/bin/psql")
.arg("-U")
.arg(&cfg.username)
.arg("-h")
.arg(&cfg.host)
.arg("-p")
.arg(cfg.port.to_string())
.arg("-d")
.arg(&cfg.database)
.arg("-t") // only return value
.arg("-c")
.arg("SHOW server_version;")
.env("PGPASSWORD", &cfg.password)
.output()?;
if !output.status.success() {
anyhow::bail!("Failed to get PostgreSQL version");
}
let version_str = String::from_utf8_lossy(&output.stdout).trim().to_string();
let major_version: u32 = version_str
.split('.')
.next()
.ok_or_else(|| anyhow::anyhow!("Cannot parse PostgreSQL version"))?
.parse()
.map_err(|_| anyhow::anyhow!("Failed to parse PostgreSQL major version"))?;
if !(12..=18).contains(&major_version) {
anyhow::bail!(
"PostgreSQL version {} not supported, must be between 12 and 18",
major_version
);
}
Ok(version_str)
}
fn select_pg_path(version: &str) -> PathBuf {
let major = version.split('.').next().unwrap_or("17"); // default to 17
PathBuf::from(format!("/usr/lib/postgresql/{}/bin", major))
}
}
#[async_trait]
impl Database for PostgresDatabase {
fn file_extension(&self) -> &'static str {
match self.format {
PostgresDumpFormat::Fc => ".dump",
PostgresDumpFormat::Fd => ".tar.gz",
}
}
async fn ping(&self) -> Result<bool> {
let server_version = PostgresDatabase::get_postgres_server_version(&self.cfg)?;
let pg_path = PostgresDatabase::select_pg_path(&server_version);
let pg_isready_path = format!("{}/pg_isready", pg_path.display());
debug!("Server version: {}", server_version);
debug!("pg_isready_path: {}", pg_isready_path);
let url = format!(
"postgresql://{}:{}@{}:{}/{}",
self.cfg.username, self.cfg.password, self.cfg.host, self.cfg.port, self.cfg.database
);
let status = Command::new(pg_isready_path)
.arg("--dbname")
.arg(url)
.status()
.context("Failed to ping Postgres")?;
Ok(status.success())
}
async fn backup(&self, backup_dir: &Path) -> Result<PathBuf> {
let server_version = PostgresDatabase::get_postgres_server_version(&self.cfg)?;
let pg_path = PostgresDatabase::select_pg_path(&server_version);
let pg_dump_path = format!("{}/pg_dump", pg_path.display());
debug!("Server version: {}", server_version);
debug!("pg_dump_path: {}", pg_dump_path);
match self.format {
PostgresDumpFormat::Fc => {
let file_path = backup_dir.join(format!(
"{}{}",
self.cfg.generated_id,
self.file_extension()
));
let url = format!(
"postgresql://{}:{}@{}:{}/{}",
self.cfg.username,
self.cfg.password,
self.cfg.host,
self.cfg.port,
self.cfg.database
);
let status = Command::new(pg_dump_path)
.arg("--dbname")
.arg(url)
.arg("-Fc")
.arg("-f")
.arg(&file_path)
.arg("-v")
.arg("--compress=3")
.status()?;
if !status.success() {
anyhow::bail!("Postgres backup failed for {}", self.cfg.name);
}
Ok(file_path)
}
PostgresDumpFormat::Fd => {
// directory dump -> tar.gz
let dump_dir = backup_dir.join(format!("{}_dir", self.cfg.generated_id));
let tar_file = backup_dir.join(format!("{}.tar.gz", self.cfg.generated_id));
std::fs::create_dir_all(&dump_dir)?;
let url = format!(
"postgresql://{}:{}@{}:{}/{}",
self.cfg.username,
self.cfg.password,
self.cfg.host,
self.cfg.port,
self.cfg.database
);
let status = Command::new("pg_dump")
.arg("--dbname")
.arg(url)
.arg("-Fd")
.arg("-j")
.arg("4")
.arg("-f")
.arg(&dump_dir)
.arg("-v")
.status()?;
if !status.success() {
anyhow::bail!("Postgres Fd backup failed for {}", self.cfg.name);
}
// Compression tar.gz
let tar_gz = std::fs::File::create(&tar_file)?;
let enc = GzEncoder::new(tar_gz, Compression::default());
let mut tar = tar::Builder::new(enc);
tar.append_dir_all(".", &dump_dir)?;
tar.finish()?;
Ok(tar_file)
}
}
}
async fn restore(&self, restore_file: &Path) -> Result<()> {
let server_version = PostgresDatabase::get_postgres_server_version(&self.cfg)?;
let pg_path = PostgresDatabase::select_pg_path(&server_version);
let pg_restore_path = format!("{}/pg_restore", pg_path.display());
let psql_path = format!("{}/psql", pg_path.display());
debug!("Server version: {}", server_version);
debug!("pg_restore_path: {}", pg_restore_path);
debug!("psql_path: {}", psql_path);
let url = format!(
"postgresql://{}:{}@{}:{}/{}",
self.cfg.username, self.cfg.password, self.cfg.host, self.cfg.port, "postgres"
);
// Terminate connections
let terminate_cmd = format!(
"SELECT pg_terminate_backend(pid) FROM pg_stat_activity WHERE datname='{}' AND pid<>pg_backend_pid();",
self.cfg.database
);
Command::new(psql_path)
.arg("-U")
.arg(&self.cfg.username)
.arg("-d")
.arg("postgres")
.arg("-h")
.arg(&self.cfg.host)
.arg("-p")
.arg(self.cfg.port.to_string())
.arg("-c")
.arg(&terminate_cmd)
.env("PGPASSWORD", &self.cfg.password)
.status()?;
match self.format {
PostgresDumpFormat::Fc => {
let status = Command::new(pg_restore_path)
.arg("--no-owner")
.arg("--no-privileges")
.arg("--clean")
.arg("--if-exists")
.arg("--create")
.arg("--dbname")
.arg(url)
.arg("-v")
.arg(restore_file)
.env("PGPASSWORD", &self.cfg.password)
.status()?;
if !status.success() {
anyhow::bail!("Postgres restore failed for {}", self.cfg.name);
}
}
PostgresDumpFormat::Fd => {
let tar_gz = std::fs::File::open(restore_file)?;
let dec = flate2::read::GzDecoder::new(tar_gz);
let mut archive = tar::Archive::new(dec);
let tmp_dir = tempfile::TempDir::new()?;
archive.unpack(tmp_dir.path())?;
let dump_dir = tmp_dir.path();
info!("Restoring dump from {}", dump_dir.display());
let status = Command::new(pg_restore_path)
.arg("--no-owner")
.arg("--no-privileges")
.arg("--clean")
.arg("--if-exists")
.arg("--create")
.arg("--dbname")
.arg(url)
.arg("-v")
.arg(dump_dir)
.env("PGPASSWORD", &self.cfg.password)
.status()?;
if !status.success() {
anyhow::bail!("Postgres Fd restore failed for {}", self.cfg.name);
}
}
}
Ok(())
}
}
+144
View File
@@ -0,0 +1,144 @@
use anyhow::Result;
use tracing::{debug, error, info};
use std::path::PathBuf;
use std::process::Command;
use super::connection::{select_pg_path, server_version};
use super::format::PostgresDumpFormat;
use crate::services::config::DatabaseConfig;
pub async fn run(
cfg: DatabaseConfig,
format: PostgresDumpFormat,
backup_dir: PathBuf,
) -> Result<PathBuf> {
tokio::task::spawn_blocking(move || -> Result<PathBuf> {
debug!("Starting backup for database {}", cfg.name);
let version = match futures::executor::block_on(server_version(&cfg)) {
Ok(v) => {
debug!("Postgres version detected: {}", v);
v
}
Err(e) => {
error!("Failed to get server version for {}: {:?}", cfg.name, e);
return Err(e.into());
}
};
let pg_dump = select_pg_path(&version).join("pg_dump");
debug!("Using pg_dump at {:?}", pg_dump);
match format {
PostgresDumpFormat::Fc => {
info!("Running FC backup for {}", cfg.name);
let file_path = backup_dir.join(format!("{}.dump", cfg.generated_id));
let url = format!(
"postgresql://{}:{}@{}:{}/{}",
cfg.username, cfg.password, cfg.host, cfg.port, cfg.database
);
let status = Command::new(&pg_dump)
.arg("--dbname")
.arg(&url)
.arg("-Fc")
.arg("-f")
.arg(&file_path)
.arg("-v")
.arg("--compress=3")
.status();
match status {
Ok(s) if s.success() => info!(
"FC backup completed successfully for {} at {:?}",
cfg.name, file_path
),
Ok(s) => {
error!("FC backup failed with status {:?} for {}", s, cfg.name);
anyhow::bail!("Postgres backup failed for {}", cfg.name);
}
Err(e) => {
error!("Error executing pg_dump for {}: {:?}", cfg.name, e);
return Err(e.into());
}
}
info!("Backup finished for database {}", cfg.name);
Ok(file_path)
}
PostgresDumpFormat::Fd => {
info!("Running FD backup for {}", cfg.name);
let dump_dir = backup_dir.join(format!("{}_dir", cfg.generated_id));
let tar_file = backup_dir.join(format!("{}.tar.gz", cfg.generated_id));
if let Err(e) = std::fs::create_dir_all(&dump_dir) {
error!(
"Failed to create dump directory {:?} for {}: {:?}",
dump_dir, cfg.name, e
);
return Err(e.into());
}
let url = format!(
"postgresql://{}:{}@{}:{}/{}",
cfg.username, cfg.password, cfg.host, cfg.port, cfg.database
);
let status = Command::new(&pg_dump)
.arg("--dbname")
.arg(&url)
.arg("-Fd")
.arg("-j")
.arg("4")
.arg("-f")
.arg(&dump_dir)
.arg("-v")
.status();
match status {
Ok(s) if s.success() => {
info!("FD backup pg_dump completed successfully for {}", cfg.name)
}
Ok(s) => {
error!(
"FD backup pg_dump failed with status {:?} for {}",
s, cfg.name
);
anyhow::bail!("Postgres FD backup failed for {}", cfg.name);
}
Err(e) => {
error!("Error executing pg_dump for {}: {:?}", cfg.name, e);
return Err(e.into());
}
}
match std::fs::File::create(&tar_file) {
Ok(tar_gz) => {
let enc =
flate2::write::GzEncoder::new(tar_gz, flate2::Compression::default());
let mut tar = tar::Builder::new(enc);
if let Err(e) = tar.append_dir_all(".", &dump_dir) {
error!("Failed to append dump_dir to tar for {}: {:?}", cfg.name, e);
return Err(e.into());
}
if let Err(e) = tar.finish() {
error!("Failed to finish tar archive for {}: {:?}", cfg.name, e);
return Err(e.into());
}
info!("FD backup archive created at {:?}", tar_file);
}
Err(e) => {
error!(
"Failed to create tar.gz file {:?} for {}: {:?}",
tar_file, cfg.name, e
);
return Err(e.into());
}
}
info!("Backup finished for database {}", cfg.name);
Ok(tar_file)
}
}
})
.await?
}
+94
View File
@@ -0,0 +1,94 @@
use std::path::Path;
use crate::domain::postgres::format::PostgresDumpFormat;
use crate::services::config::DatabaseConfig;
use anyhow::Result;
use tokio_postgres::{Client, NoTls};
use tracing::info;
pub async fn connect(cfg: &DatabaseConfig) -> Result<Client> {
let dsn = format!(
"host={} port={} user={} password={} dbname={}",
cfg.host, cfg.port, cfg.username, cfg.password, cfg.database
);
let (client, connection) = tokio_postgres::connect(&dsn, NoTls).await?;
tokio::spawn(async move {
if let Err(e) = connection.await {
tracing::error!("Postgres connection error: {}", e);
}
});
Ok(client)
}
pub async fn server_version(cfg: &DatabaseConfig) -> Result<String> {
let client = connect(cfg).await?;
let version: String = client.query_one("SHOW server_version;", &[]).await?.get(0);
Ok(version)
}
pub fn select_pg_path(version: &str) -> std::path::PathBuf {
let major = version.split('.').next().unwrap_or("17");
format!("/usr/lib/postgresql/{}/bin", major).into()
}
pub async fn terminate_connections(cfg: &DatabaseConfig) -> Result<()> {
let mut admin = cfg.clone();
admin.database = "postgres".into();
let client = connect(&admin).await?;
client
.execute(
r#"
SELECT pg_terminate_backend(pid)
FROM pg_stat_activity
WHERE datname = $1
AND pid <> pg_backend_pid();
"#,
&[&cfg.database],
)
.await?;
Ok(())
}
pub fn detect_format_from_file(restore_file: &Path) -> PostgresDumpFormat {
match restore_file.extension().and_then(|e| e.to_str()) {
Some("dump") => PostgresDumpFormat::Fc,
Some("gz") => PostgresDumpFormat::Fd,
// Some("tar.gz") => PostgresDumpFormat::Fd,
_ => PostgresDumpFormat::Fc,
}
}
pub async fn detect_format_from_size(cfg: &DatabaseConfig) -> PostgresDumpFormat {
info!(
"Detecting database format {:?} - {:?}",
cfg.name, cfg.generated_id
);
let client = match connect(cfg).await {
Ok(c) => c,
Err(_) => return PostgresDumpFormat::Fc,
};
let row = match client
.query_one("SELECT pg_database_size(current_database());", &[])
.await
{
Ok(r) => r,
Err(_) => return PostgresDumpFormat::Fc,
};
let size_bytes: i64 = row.get(0);
info!("Size of database is {} bytes", size_bytes);
// > 1 Go
if size_bytes > 1_000_000_000 {
info!("Using -Fd format");
PostgresDumpFormat::Fd
} else {
info!("Using -Fc format");
PostgresDumpFormat::Fc
}
}
+53
View File
@@ -0,0 +1,53 @@
use anyhow::Result;
use async_trait::async_trait;
use std::path::{Path, PathBuf};
use super::{
backup,
format::PostgresDumpFormat,
ping, restore,
};
use crate::domain::factory::Database;
use crate::services::config::DatabaseConfig;
use crate::utils::locks::{DbOpLock, FileLock};
pub struct PostgresDatabase {
pub cfg: DatabaseConfig,
pub format: PostgresDumpFormat,
}
impl PostgresDatabase {
pub fn new(cfg: DatabaseConfig, format: PostgresDumpFormat) -> Self {
Self { cfg, format }
}
}
#[async_trait]
impl Database for PostgresDatabase {
fn file_extension(&self) -> &'static str {
match self.format {
PostgresDumpFormat::Fc => ".dump",
PostgresDumpFormat::Fd => ".gz",
// PostgresDumpFormat::Fd => ".tar.gz",
}
}
async fn ping(&self) -> Result<bool> {
ping::run(self.cfg.clone()).await
}
async fn backup(&self, dir: &Path) -> Result<PathBuf> {
FileLock::acquire(&self.cfg.generated_id, DbOpLock::Backup.as_str()).await?;
let res = backup::run(self.cfg.clone(), self.format, dir.to_path_buf()).await;
FileLock::release(&self.cfg.generated_id).await?;
res
}
async fn restore(&self, file: &Path) -> Result<()> {
FileLock::acquire(&self.cfg.generated_id, DbOpLock::Restore.as_str()).await?;
let res = restore::run(self.cfg.clone(), self.format, file.to_path_buf()).await;
FileLock::release(&self.cfg.generated_id).await?;
res
}
}
+5
View File
@@ -0,0 +1,5 @@
#[derive(Clone, Copy)]
pub enum PostgresDumpFormat {
Fc,
Fd,
}
+8
View File
@@ -0,0 +1,8 @@
pub mod backup;
pub mod database;
mod restore;
mod connection;
mod format;
mod ping;
pub use connection::{detect_format_from_size, detect_format_from_file};
+8
View File
@@ -0,0 +1,8 @@
use super::connection::connect;
use crate::services::config::DatabaseConfig;
pub async fn run(
cfg: DatabaseConfig,
) -> anyhow::Result<bool> {
Ok(connect(&cfg).await.is_ok())
}
+168
View File
@@ -0,0 +1,168 @@
use anyhow::Result;
use tracing::{debug, error, info};
use std::path::PathBuf;
use std::process::Command;
use super::connection::{select_pg_path, server_version, terminate_connections};
use super::format::PostgresDumpFormat;
use crate::services::config::DatabaseConfig;
pub async fn run(
cfg: DatabaseConfig,
format: PostgresDumpFormat,
restore_file: PathBuf,
) -> Result<()> {
tokio::task::spawn_blocking(move || -> Result<()> {
debug!("Starting restore for database {}", cfg.name);
let version = match futures::executor::block_on(server_version(&cfg)) {
Ok(v) => {
debug!("Postgres version detected: {}", v);
v
}
Err(e) => {
error!("Failed to get server version for {}: {:?}", cfg.name, e);
return Err(e.into());
}
};
let pg_restore = select_pg_path(&version).join("pg_restore");
debug!("Using pg_restore at {:?}", pg_restore);
if let Err(e) = futures::executor::block_on(terminate_connections(&cfg)) {
error!("Failed to terminate connections for {}: {:?}", cfg.name, e);
return Err(e.into());
}
info!("Connections terminated for database {}", cfg.name);
let url = format!(
"postgresql://{}:{}@{}:{}/postgres",
cfg.username, cfg.password, cfg.host, cfg.port
);
debug!("Restore URL: {}", url);
match format {
PostgresDumpFormat::Fc => {
info!("Running FC restore for {}", cfg.name);
let status = Command::new(&pg_restore)
.arg("--no-owner")
.arg("--no-privileges")
.arg("--clean")
.arg("--if-exists")
.arg("--create")
.arg("--dbname")
.arg(&url)
.arg("-v")
.arg(&restore_file)
.env("PGPASSWORD", &cfg.password)
.status();
match status {
Ok(s) if s.success() => {
info!("FC restore completed successfully for {}", cfg.name)
}
Ok(s) => {
error!("FC restore failed with status {:?} for {}", s, cfg.name);
anyhow::bail!("Postgres restore failed for {}", cfg.name);
}
Err(e) => {
error!("Error executing pg_restore for {}: {:?}", cfg.name, e);
return Err(e.into());
}
}
}
PostgresDumpFormat::Fd => {
info!("Running FD restore for {}", cfg.name);
let tar_gz = match std::fs::File::open(&restore_file) {
Ok(f) => f,
Err(e) => {
error!(
"Failed to open restore file {:?} for {}: {:?}",
restore_file, cfg.name, e
);
return Err(e.into());
}
};
let dec = flate2::read::GzDecoder::new(tar_gz);
let mut archive = tar::Archive::new(dec);
let tmp_dir = match tempfile::TempDir::new() {
Ok(d) => d,
Err(e) => {
error!(
"Failed to create temporary directory for FD restore of {}: {:?}",
cfg.name, e
);
return Err(e.into());
}
};
if let Err(e) = archive.unpack(tmp_dir.path()) {
error!("Failed to unpack FD archive for {}: {:?}", cfg.name, e);
return Err(e.into());
}
debug!("Listing contents of temp dir: {}", tmp_dir.path().display());
for entry in std::fs::read_dir(tmp_dir.path())? {
if let Ok(entry) = entry {
let path = entry.path();
let file_type = entry.file_type()?;
debug!(
" - {} | is_dir: {} | is_file: {}",
path.display(),
file_type.is_dir(),
file_type.is_file()
);
}
}
let dump_dir = if tmp_dir.path().join("toc.dat").exists() {
tmp_dir.path().to_path_buf()
} else {
std::fs::read_dir(tmp_dir.path())?
.filter_map(|e| e.ok())
.find(|entry| entry.path().join("toc.dat").exists())
.map(|e| e.path())
.ok_or_else(|| anyhow::anyhow!("Invalid FD archive: toc.dat not found"))?
};
let status = Command::new(&pg_restore)
.arg("--no-owner")
.arg("--no-privileges")
.arg("--clean")
.arg("--if-exists")
.arg("--create")
.arg("--dbname")
.arg(&url)
.arg("-v")
.arg("-j")
.arg("4")
.arg(dump_dir)
.env("PGPASSWORD", &cfg.password)
.status();
match status {
Ok(s) if s.success() => {
info!("FD restore completed successfully for {}", cfg.name)
}
Ok(s) => {
error!("FD restore failed with status {:?} for {}", s, cfg.name);
anyhow::bail!("Postgres FD restore failed for {}", cfg.name);
}
Err(e) => {
error!("Error executing pg_restore for {}: {:?}", cfg.name, e);
return Err(e.into());
}
}
}
}
info!("Restore finished for database {}", cfg.name);
Ok(())
})
.await?
}
+17 -13
View File
@@ -1,24 +1,28 @@
mod core;
mod domain;
mod services;
mod settings;
mod tasks;
mod utils;
mod core;
mod services;
mod domain;
use tracing_subscriber;
use crate::tasks::ping::ping_server;
use crate::utils::locks::FileLock;
use utils::redis_client;
use utils::task_manager::scheduler;
use crate::tasks::ping::ping_server;
use crate::utils::logging;
#[tokio::main]
async fn main() {
tracing_subscriber::fmt::init();
tokio::join!(
ping_server(),
async {
let conn = redis_client::redis_connection().await;
scheduler::scheduler_loop(conn).await;
}
);
logging::init_logger();
// Remove all locks on startup
if let Err(e) = FileLock::clean_startup().await {
eprintln!("Failed to clean locks on startup: {:?}", e);
}
tokio::join!(ping_server(), async {
let conn = redis_client::redis_connection().await;
scheduler::scheduler_loop(conn).await;
});
}
+34 -10
View File
@@ -2,12 +2,11 @@
use crate::core::context::Context;
use crate::domain::factory::DatabaseFactory;
use crate::services::config::{DatabaseConfig, DatabasesConfig};
use crate::services::config::{DatabaseConfig, DatabasesConfig, DbType};
use crate::utils::common::BackupMethod;
use crate::utils::file::full_extension;
use anyhow::Result;
use hex;
use log::{error, info};
use openssl::encrypt::Encrypter;
use openssl::hash::MessageDigest;
use openssl::pkey::PKey;
@@ -19,13 +18,15 @@ use std::path::{Path, PathBuf};
use std::sync::Arc;
use tempfile::TempDir;
use tokio::fs;
use tracing::{error, info};
#[derive(Debug)]
pub struct BackupResult {
pub generated_id: String,
pub db_type: String,
pub db_type: DbType,
pub status: String,
pub backup_file: Option<PathBuf>,
pub code: Option<String>,
}
pub struct BackupService {
@@ -85,6 +86,7 @@ impl BackupService {
db_type,
status: "failed".into(),
backup_file: None,
code: None,
});
}
@@ -94,20 +96,42 @@ impl BackupService {
db_type,
status: "success".into(),
backup_file: Some(file),
code: None,
}),
Err(_) => Ok(BackupResult {
generated_id,
db_type,
status: "failed".into(),
backup_file: None,
}),
Err(e) => match e.to_string().as_str() {
"backup_already_in_progress" => Ok(BackupResult {
generated_id,
db_type,
status: "failed".into(),
backup_file: None,
code: Some(e.to_string()),
}),
_ => Ok(BackupResult {
generated_id,
db_type,
status: "failed".into(),
backup_file: None,
code: None,
}),
},
}
}
pub async fn send_result(&self, result: BackupResult, method: BackupMethod) {
if result.code.as_deref() == Some("backup_already_in_progress") {
info!(
"[BackupService] Skipping send for DB {}: backup already in progress",
result.generated_id
);
return;
}
info!(
"[BackupService] DB: {} Type: {} Status: {} File: {:?}",
result.generated_id, result.db_type, result.status, result.backup_file
result.generated_id,
result.db_type.as_str(),
result.status,
result.backup_file
);
let client = reqwest::Client::new();
+28 -2
View File
@@ -10,13 +10,36 @@ use std::sync::Arc;
use toml;
use tracing::info;
#[derive(Debug, Deserialize, Clone)]
#[serde(rename_all = "lowercase")]
pub enum DbType {
Mysql,
Mariadb,
Postgresql,
MongoDB
// Sqlite,
// Add other DB types if needed
}
impl DbType {
pub fn as_str(&self) -> &'static str {
match self {
DbType::Mysql => "mysql",
DbType::Mariadb => "mysql",
DbType::Postgresql => "postgresql",
DbType::MongoDB => "mongodb",
// DbType::Sqlite => "sqlite",
}
}
}
#[allow(dead_code)]
#[derive(Debug, Deserialize, Clone)]
pub struct DatabaseConfig {
pub name: String,
pub database: String,
#[serde(rename = "type")]
pub db_type: String,
pub db_type: DbType,
pub username: String,
pub password: String,
pub port: u16,
@@ -55,7 +78,10 @@ impl ConfigService {
let path_obj = Path::new(&path);
if !path_obj.exists() {
return Err(format!("Config file not found: {}", &path));
return Err(format!(
"Config file not found: {}, check documentation and add config file.",
&path
));
}
let extension = path_obj
+4 -1
View File
@@ -5,7 +5,7 @@ use crate::domain::factory::DatabaseFactory;
use crate::services::config::{DatabaseConfig, DatabasesConfig};
use crate::services::status::DatabaseStatus;
use anyhow::Result;
use log::{error, info};
use tracing::{error, info};
use serde::Serialize;
use std::path::Path;
use std::sync::Arc;
@@ -65,6 +65,8 @@ impl RestoreService {
) -> Result<RestoreResult> {
let generated_id = cfg.generated_id.clone();
info!("File url: {}", file_url);
let client = reqwest::Client::new();
let response = client.get(file_url).send().await?;
if !response.status().is_success() {
@@ -99,6 +101,7 @@ impl RestoreService {
let db_instance = DatabaseFactory::create_for_restore(cfg.clone(), &backup_file_path).await;
let reachable = db_instance.ping().await.unwrap_or(false);
info!("Reachable: {}", reachable);
if !reachable {
return Ok(RestoreResult {
generated_id,
+2 -2
View File
@@ -86,12 +86,12 @@ impl StatusService {
.iter()
.map(|db| DatabasePayload {
name: &db.name,
dbms: &db.db_type,
dbms: &db.db_type.as_str(),
generated_id: &db.generated_id,
})
.collect();
let version_str = format!("{}-rust", CONFIG.app_version);
let version_str = CONFIG.app_version.as_str();
let body = StatusRequestBody {
version: &version_str,
+14 -1
View File
@@ -11,12 +11,22 @@ pub struct Settings {
pub edge_key: String,
pub databases_config_file: String,
pub data_path: String,
pub pooling: usize,
pub timezone: String,
pub log: String,
}
impl Settings {
fn from_env() -> Self {
dotenv().ok();
let pooling_seconds = env::var("POOLING")
.unwrap_or_else(|_| "5".to_string())
.parse::<usize>()
.expect("POOLING must be a valid positive integer");
let tz = env::var("TZ").unwrap_or_else(|_| "UTC".to_string());
Self {
app_version: env!("CARGO_PKG_VERSION").to_string(),
app_env: env::var("APP_ENV").unwrap_or_else(|_| "development".into()),
@@ -24,8 +34,11 @@ impl Settings {
.unwrap_or_else(|_| "redis://localhost:6379/".into()),
edge_key: env::var("EDGE_KEY").unwrap_or_default(),
databases_config_file: env::var("DATABASES_CONFIG_FILE")
.unwrap_or_else(|_| "config.toml".into()),
.unwrap_or_else(|_| "config.json".into()),
data_path: env::var("DATA_PATH").unwrap_or_else(|_| "/config".into()),
pooling: pooling_seconds,
timezone: tz,
log: env::var("LOG").unwrap_or_else(|_| "info".into()),
}
}
}
+28 -1
View File
@@ -1,5 +1,6 @@
use crate::core::agent::Agent;
use crate::core::context::Context;
use crate::settings::CONFIG;
use crate::utils::common::BackupMethod;
use std::sync::Arc;
use std::time::Duration;
@@ -19,6 +20,32 @@ pub async fn ping_server() {
);
}
tokio::time::sleep(Duration::from_secs(60)).await;
tokio::time::sleep(Duration::from_secs(CONFIG.pooling as u64)).await;
}
}
// use crate::core::agent::Agent;
// use crate::core::context::Context;
// use crate::utils::common::BackupMethod;
// use std::sync::Arc;
// use tokio::sync::Mutex;
// use tokio::time::{sleep, Duration};
// use tracing::{error};
//
// pub async fn ping_server() {
//
//
// loop {
// let ctx = Arc::new(Context::new());
// let agent = Arc::new(Mutex::new(Agent::new(ctx.clone()).await));
// let agent_clone = agent.clone();
//
// tokio::spawn(async move {
// let mut agent_locked = agent_clone.lock().await;
// if let Err(e) = agent_locked.run(BackupMethod::Manual).await {
// error!("An error occurred while executing ping_server: {:?}", e);
// }
// });
//
// sleep(Duration::from_secs(5)).await;
// }
// }
+113
View File
@@ -0,0 +1,113 @@
use anyhow::{Context, Result};
use chrono::{Local};
use tracing::{info, warn, error};
use std::path::{Path, PathBuf};
use std::time::{Duration, SystemTime};
use tokio::fs::{OpenOptions, metadata, remove_file, create_dir_all, read_dir};
use tokio::io::AsyncWriteExt;
/// Lock type for logging purposes
#[derive(Debug, Copy, Clone)]
pub enum DbOpLock {
Backup,
Restore,
}
impl DbOpLock {
/// Convert enum to a static string for service name
pub fn as_str(&self) -> &'static str {
match self {
DbOpLock::Backup => "backup-service",
DbOpLock::Restore => "restore-service",
}
}
}
/// File-based lock utility
pub struct FileLock;
impl FileLock {
const LOCK_DIR: &'static str = "/var/locks";
/// Returns the path for the lock file
fn lock_file_path(id: &str) -> PathBuf {
Path::new(Self::LOCK_DIR).join(format!("{}.lock", id))
}
/// Ensure the locks directory exists
async fn ensure_lock_dir() -> Result<()> {
create_dir_all(Self::LOCK_DIR)
.await
.with_context(|| format!("Failed to create locks directory {}", Self::LOCK_DIR))?;
Ok(())
}
/// Clean all lock files on startup (remove everything, ignore stale)
pub async fn clean_startup() -> Result<()> {
Self::ensure_lock_dir().await?;
let mut dir = read_dir(Self::LOCK_DIR).await?;
while let Some(entry) = dir.next_entry().await? {
let path = entry.path();
if path.is_file() && path.extension().map(|e| e == "lock").unwrap_or(false) {
if let Err(e) = remove_file(&path).await {
warn!("Failed to remove lock file {:?}: {:?}", path, e);
} else {
info!("Removed lock file {:?}", path);
}
}
}
Ok(())
}
/// Acquire a file-based lock
pub async fn acquire(id: &str, service_name: &str) -> Result<()> {
Self::ensure_lock_dir().await?;
let path = Self::lock_file_path(id);
info!("Attempting to acquire lock for {} at {:?}", id, path);
if path.exists() {
let meta = metadata(&path).await?;
if let Ok(modified) = meta.modified() {
let age = SystemTime::now().duration_since(modified)?;
if age > Duration::from_secs(24 * 60 * 60) {
remove_file(&path).await?;
warn!("Removed stale lock for {}", id);
} else {
error!("Lock already held for {}. Cannot acquire.", id);
anyhow::bail!("backup_already_in_progress");
}
}
}
let mut f = OpenOptions::new()
.write(true)
.create_new(true)
.open(&path)
.await
.with_context(|| format!("Failed to create lock file for {}", id))?;
f.write_all(format!("Service: {}\n", service_name).as_bytes()).await?;
f.write_all(format!("PID: {}\n", std::process::id()).as_bytes()).await?;
// f.write_all(format!("Timestamp: {}\n", Utc::now()).as_bytes()).await?;
f.write_all(format!("Timestamp: {}\n", Local::now()).as_bytes()).await?;
info!("Successfully acquired lock for {}", id);
Ok(())
}
/// Release the file-based lock
pub async fn release(id: &str) -> Result<()> {
let path = Self::lock_file_path(id);
info!("Releasing lock for {}", id);
if path.exists() {
remove_file(&path).await?;
info!("Released file lock for {}", id);
} else {
warn!("Attempted to release lock for {}, but file does not exist", id);
}
Ok(())
}
}
+52
View File
@@ -0,0 +1,52 @@
use crate::settings::CONFIG;
use once_cell::sync::Lazy;
use time::macros::format_description;
use tracing_appender::non_blocking::{NonBlocking, WorkerGuard};
use tracing_appender::rolling::{RollingFileAppender, Rotation};
use tracing_subscriber::fmt::time::LocalTime;
use tracing_subscriber::{EnvFilter, Registry, fmt, prelude::*};
const LOGS_DIRECTORY: &str = "/var/log/app";
static FILE_APPENDER: Lazy<(NonBlocking, WorkerGuard)> = Lazy::new(|| {
let file_appender = RollingFileAppender::new(Rotation::DAILY, LOGS_DIRECTORY, "app.log");
tracing_appender::non_blocking(file_appender)
});
pub fn init_logger() {
std::fs::create_dir_all(LOGS_DIRECTORY).ok();
let (writer, _guard) = &*FILE_APPENDER;
let timer = LocalTime::new(format_description!(
"[year]-[month]-[day]T[hour]:[minute]:[second]"
));
let file_layer = fmt::layer()
.with_writer(writer.clone())
.with_timer(timer.clone())
.with_ansi(false)
.with_target(false);
// let env_filter = EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new("info"));
let env_filter = EnvFilter::new(CONFIG.log.clone());
let term_layer = fmt::layer()
.with_writer(std::io::stdout)
.with_timer(timer.clone())
// .with_timer(tracing_subscriber::fmt::time::LocalTime::rfc_3339())
.with_ansi(true)
.with_target(false)
.with_filter(env_filter);
let subscriber = Registry::default().with(file_layer).with(term_layer);
tracing::subscriber::set_global_default(subscriber).expect("Failed to set global subscriber");
tracing::info!(
"Logger initialized | TZ={} | POOLING={}s",
CONFIG.timezone,
CONFIG.pooling
);
}
+3 -1
View File
@@ -3,4 +3,6 @@ pub mod edge_key;
pub mod redis_client;
pub mod task_manager;
pub mod text;
pub mod file;
pub mod file;
pub mod locks;
pub mod logging;
-10
View File
@@ -1,13 +1,3 @@
// use redis::{Client, Connection};
// use crate::settings::CONFIG;
//
// pub fn redis_connection() -> Connection {
// let client = Client::open(CONFIG.redis_url.clone()).expect("Invalid Redis URL");
// client.get_connection().expect("Failed to connect to Redis")
//
// }
use redis::{aio::MultiplexedConnection, Client};
use crate::settings::CONFIG;
+5 -71
View File
@@ -1,84 +1,18 @@
// use crate::utils::task_manager::models;
// use crate::utils::task_manager::tasks::{remove_task, upsert_task};
// use crate::utils::text::normalize_cron;
// use chrono::Utc;
// use cron::Schedule;
// use log::debug;
// use redis::Commands;
// use std::str::FromStr;
// use tracing::info;
//
// pub fn next_run_timestamp(expr: &str) -> i64 {
// let schedule = Schedule::from_str(expr).unwrap();
// schedule.upcoming(Utc).next().unwrap().timestamp()
// }
//
// pub fn check_and_update_cron(
// conn: &mut redis::aio::MultiplexedConnection,
// cron_value: Option<String>,
// args: Vec<String>,
// task: &str,
// task_name: String,
// ) {
// let redis_key = format!("redbeat:{}", task_name);
//
// let exists: bool = conn.exists(&redis_key).unwrap();
//
// match cron_value {
// None => {
// if exists {
// remove_task(conn, &task_name).unwrap();
// info!("Task {} removed", task_name);
// }
// }
//
// Some(cron) => {
// let cron = normalize_cron(&cron);
// debug!("Task cron (normalized): {:?}", cron);
//
// if exists {
// let raw: String = conn.hget(&redis_key, "data").unwrap();
// let stored: models::PeriodicTask = serde_json::from_str(&raw).unwrap();
//
// if stored.cron != cron {
// upsert_task(
// conn,
// &task_name,
// task,
// &cron,
// args,
// )
// .unwrap();
// info!("Task {} updated", task_name);
// }
// } else {
// upsert_task(
// conn,
// &task_name,
// task,
// &cron,
// args,
// )
// .unwrap();
// info!("Task {} created", task_name);
// }
// }
// }
// }
use crate::utils::task_manager::models;
use crate::utils::task_manager::tasks::{remove_task, upsert_task};
use crate::utils::text::normalize_cron;
use chrono::Utc;
use chrono::{Local};
use cron::Schedule;
use log::debug;
use redis::aio::MultiplexedConnection;
use tracing::debug;
use redis::AsyncCommands;
use redis::aio::MultiplexedConnection;
use std::str::FromStr;
use tracing::info;
pub fn next_run_timestamp(expr: &str) -> i64 {
let schedule = Schedule::from_str(expr).unwrap();
schedule.upcoming(Utc).next().unwrap().timestamp()
// schedule.upcoming(Utc).next().unwrap().timestamp()
schedule.upcoming(Local).next().unwrap().timestamp()
}
pub async fn check_and_update_cron(
+13 -3
View File
@@ -5,14 +5,17 @@ use crate::utils::common::BackupMethod;
use crate::utils::task_manager::cron::next_run_timestamp;
use crate::utils::task_manager::models::PeriodicTask;
use crate::utils::task_manager::tasks::SCHEDULE_KEY;
use log::info;
use tracing::info;
use redis::AsyncCommands;
use redis::aio::MultiplexedConnection;
use std::sync::Arc;
use tracing::error;
pub async fn scheduler_loop(mut conn: MultiplexedConnection) {
loop {
let now = chrono::Utc::now().timestamp();
// let now = chrono::Utc::now().timestamp();
let now = chrono::Local::now().timestamp();
// info!("Scheduling task {}", chrono::Local::now());
let due: Vec<String> = conn
.zrangebyscore(SCHEDULE_KEY, 0, now)
@@ -36,7 +39,14 @@ pub async fn scheduler_loop(mut conn: MultiplexedConnection) {
task_clone.task, task_clone.args
);
let _ = execute_task(task_clone.task.as_str(), task_clone.args).await;
// let _ = execute_task(task_clone.task.as_str(), task_clone.args).await;
if let Err(e) = execute_task(task_clone.task.as_str(), task_clone.args).await {
error!(
"An error occurred while executing task={} : {:?}",
task_clone.task, e
);
}
let next_ts = next_run_timestamp(&task_clone.cron);
let _: () = conn_clone.zadd(SCHEDULE_KEY, &key, next_ts).await.unwrap();
-55
View File
@@ -1,60 +1,5 @@
#![allow(dead_code)]
// use redis::{Connection};
// use redis::aio::MultiplexedConnection;
// use crate::utils::task_manager::cron::next_run_timestamp;
// use crate::utils::task_manager::models::PeriodicTask;
//
// pub const SCHEDULE_KEY: &str = "redbeat:schedule";
//
// pub fn upsert_task(
// conn: &mut MultiplexedConnection,
// name: &str,
// task: &str,
// cron: &str,
// args: Vec<String>,
// ) -> redis::RedisResult<()> {
//
// let key = format!("redbeat:{}", name);
// let next_ts = next_run_timestamp(cron);
//
// let entry = PeriodicTask {
// task: task.to_string(),
// cron: cron.to_string(),
// args,
// enabled: true,
// };
//
// let payload = serde_json::to_string(&entry).unwrap();
//
// redis::pipe()
// .atomic()
// .cmd("HSET")
// .arg(&key)
// .arg("data")
// .arg(payload)
// .cmd("ZADD")
// .arg(SCHEDULE_KEY)
// .arg(next_ts)
// .arg(&key)
// .query(conn)
// }
//
// pub fn remove_task(
// conn: &mut MultiplexedConnection,
// name: &str,
// ) -> redis::RedisResult<()> {
// let key = format!("redbeat:{}", name);
//
// redis::pipe()
// .atomic()
// .cmd("ZREM")
// .arg(SCHEDULE_KEY)
// .arg(&key)
// .cmd("DEL")
// .arg(&key)
// .query(conn)
// }
use crate::utils::task_manager::cron::next_run_timestamp;
use crate::utils::task_manager::models::PeriodicTask;
use redis::aio::MultiplexedConnection;