import paramiko

# 創(chuàng)建 SSH 客戶端
client = paramiko.SSHClient()
client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
client.connect(hostname='your_server_ip', username='user', password='pass')

# 執(zhí)行命令
stdin, stdout, stderr = client.exec_command('ls -l /tmp')
print(stdout.read().decode())

# 關(guān)閉連接
client.close()

from fabric import Connectiondef restart_nginx():


    # 連接到服務(wù)器
    c = Connection('user@server_ip')
    # 執(zhí)行命令
    c.run('sudo systemctl restart nginx')
    print("Nginx restarted!")

- hosts: webservers  # 目標(biāo)服務(wù)器分組
  become: yes        # 使用 sudo 權(quán)限
  tasks:
    - name: Install Nginx
      apt:
        name: nginx
        state: present
    - name: Copy Config File
      copy:
        src: ./nginx.conf
        dest: /etc/nginx/nginx.conf
    - name: Start Nginx
      service:
        name: nginx
        state: restarted

ansible-playbook -i inventory.ini deploy_web.yml

salt '*' pkg.install nginx

import psutil

cpu_usage = psutil.cpu_percent(interval=1)
mem_usage = psutil.virtual_memory().percent
print(f"CPU: {cpu_usage}%, Memory: {mem_usage}%")

from prometheus_client import start_http_server, Gauge

# 定義指標(biāo)
CPU_GAUGE = Gauge('cpu_usage', 'Current CPU usage in percent')

# 啟動(dòng) HTTP 服務(wù)暴露指標(biāo)
start_http_server(8000)
while True:
    CPU_GAUGE.set(psutil.cpu_percent())

from elasticsearch import Elasticsearch

es = Elasticsearch(['http://localhost:9200'])
log_data = {
    "timestamp": "2023-10-01T12:00:00",
    "level": "ERROR",
    "message": "Disk space low on /dev/sda1"
}
es.index(index="app_logs", document=log_data)

pipeline {
    agent any
    stages {
        stage('Deploy') {
            steps {
                script {
                    sh 'python deploy.py --env production'
                }
            }
        }
    }
}

import dockerclient = docker.from_env()
container = client.containers.run(


    "nginx:latest",
    detach=True,
    ports={'80/tcp': 8080}
)
print(f"Container ID: {container.id}")

from loguru import logger

logger.add("app.log", rotation="100 MB")  # 日志文件輪轉(zhuǎn)
logger.info("Service started successfully")

from airflow import DAG
from airflow.operators.python_operator import PythonOperator
from datetime import datetime

def analyze_logs():
    print("Analyzing logs...")

dag = DAG('log_analysis', start_date=datetime(2023, 1, 1))
task = PythonOperator(
    task_id='analyze_logs',
    python_callable=analyze_logs,
    dag=dag
)

try:
    response = requests.get('http://api.example.com', timeout=5)
except requests.exceptions.Timeout:
    logger.error("API request timed out")

from apscheduler.schedulers.blocking import BlockingSchedulerscheduler = BlockingScheduler()
@scheduler.scheduled_job('interval', minutes=30)


def health_check():
    print("Performing health check...")
scheduler.start()