本文目录导读：
我来帮你写一个自动上报状态的脚本,根据不同场景，提供几种常见实现方案：
基础版 - Shell脚本 + Cron任务
#!/bin/bash
# auto_report.sh
# 配置参数
API_URL="https://your-api.com/report"
SERVER_NAME=$(hostname)
STATUS="running"
TIMESTAMP=$(date '+%Y-%m-%d %H:%M:%S')
# 收集系统状态
CPU_USAGE=$(top -bn1 | grep "Cpu(s)" | awk '{print $2}' | cut -d'%' -f1)
MEM_USAGE=$(free | grep Mem | awk '{print $3/$2 * 100.0}')
DISK_USAGE=$(df -h / | awk 'NR==2 {print $5}' | cut -d'%' -f1)
# 构建JSON数据
JSON_DATA=$(cat <<EOF
{
"server": "$SERVER_NAME",
"status": "$STATUS",
"timestamp": "$TIMESTAMP",
"metrics": {
"cpu": $CPU_USAGE,
"memory": $MEM_USAGE,
"disk": $DISK_USAGE
}
}
EOF
)
# 发送请求
curl -X POST "$API_URL" \
-H "Content-Type: application/json" \
-H "Authorization: Bearer YOUR_TOKEN" \
-d "$JSON_DATA"
# 检查结果
if [ $? -eq 0 ]; then
echo "$(date): Status reported successfully"
else
echo "$(date): Failed to report status"
fi
设置Cron定时任务：
# 每分钟执行一次 * * * * * /path/to/auto_report.sh # 每5分钟执行一次 */5 * * * * /path/to/auto_report.sh # 每天8点执行 0 8 * * * /path/to/auto_report.sh
Python版 - 更灵活
# auto_report.py
import requests
import json
import psutil
import socket
import datetime
import time
import logging
class StatusReporter:
def __init__(self, config):
self.api_url = config.get('api_url')
self.token = config.get('token')
self.interval = config.get('interval', 60) # 默认60秒
self.server_name = socket.gethostname()
# 配置日志
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s'
)
def collect_metrics(self):
"""收集系统指标"""
return {
'server': self.server_name,
'status': 'running',
'timestamp': datetime.datetime.now().isoformat(),
'metrics': {
'cpu_percent': psutil.cpu_percent(interval=1),
'memory_percent': psutil.virtual_memory().percent,
'disk_percent': psutil.disk_usage('/').percent,
'network_connections': len(psutil.net_connections())
}
}
def report_status(self):
"""上报状态"""
try:
data = self.collect_metrics()
response = requests.post(
self.api_url,
json=data,
headers={
'Authorization': f'Bearer {self.token}',
'Content-Type': 'application/json'
},
timeout=10
)
if response.status_code == 200:
logging.info(f"Status reported: {response.json()}")
return True
else:
logging.error(f"Report failed: {response.status_code}")
return False
except requests.exceptions.RequestException as e:
logging.error(f"Request failed: {e}")
return False
except Exception as e:
logging.error(f"Unexpected error: {e}")
return False
def run_continuous(self):
"""持续运行"""
logging.info(f"Starting continuous reporting every {self.interval} seconds")
while True:
self.report_status()
time.sleep(self.interval)
# 使用示例
if __name__ == "__main__":
config = {
'api_url': 'https://your-api.com/report',
'token': 'your-secret-token',
'interval': 300 # 5分钟
}
reporter = StatusReporter(config)
# 单次上报
# reporter.report_status()
# 持续运行
reporter.run_continuous()
带重试机制的可靠版本
# reliable_reporter.py
import requests
import json
import time
from retry import retry
from datetime import datetime
import hashlib
class ReliableReporter:
def __init__(self, config):
self.config = config
self.failed_attempts = 0
self.max_retries = config.get('max_retries', 3)
@retry(tries=3, delay=2, backoff=2)
def send_with_retry(self, data):
"""带重试的发送"""
response = requests.post(
self.config['api_url'],
json=data,
headers=self.get_headers(data),
timeout=5
)
response.raise_for_status()
return response
def get_headers(self, data):
"""生成请求头"""
timestamp = str(int(time.time()))
signature = hashlib.sha256(
f"{json.dumps(data)}{self.config['secret']}{timestamp}".encode()
).hexdigest()
return {
'Content-Type': 'application/json',
'X-Timestamp': timestamp,
'X-Signature': signature,
'Authorization': f'Bearer {self.config["token"]}'
}
def report_with_circuit_breaker(self):
"""带熔断机制的上报"""
if self.failed_attempts >= self.config.get('max_failures', 5):
# 进入熔断状态
print("Circuit breaker opened, waiting...")
time.sleep(self.config.get('cooldown', 300))
self.failed_attempts = 0
try:
data = self.collect_status()
self.send_with_retry(data)
self.failed_attempts = 0
return True
except Exception as e:
self.failed_attempts += 1
print(f"Report failed ({self.failed_attempts}/{self.config.get('max_failures', 5)}): {e}")
# 保存失败记录
self.save_failed_report(data)
return False
def save_failed_report(self, data):
"""保存失败的上报"""
with open('failed_reports.json', 'a') as f:
record = {
'timestamp': datetime.now().isoformat(),
'data': data,
'failed_count': self.failed_attempts
}
f.write(json.dumps(record) + '\n')
def retry_failed_reports(self):
"""重试失败的上报"""
try:
with open('failed_reports.json', 'r') as f:
for line in f:
record = json.loads(line)
try:
response = requests.post(
self.config['api_url'],
json=record['data'],
headers=self.get_headers(record['data']),
timeout=5
)
if response.status_code == 200:
print(f"Retry success for {record['timestamp']}")
except:
print(f"Retry failed for {record['timestamp']}")
except FileNotFoundError:
pass
# 配置示例
config = {
'api_url': 'https://your-api.com/report',
'token': 'your-token',
'secret': 'your-secret',
'max_retries': 3,
'max_failures': 5,
'cooldown': 300
}
reporter = ReliableReporter(config)
# 定时执行（使用schedule库）
# pip install schedule
import schedule
def job():
reporter.report_with_circuit_breaker()
# 每5分钟执行一次
schedule.every(5).minutes.do(job)
while True:
schedule.run_pending()
time.sleep(1)
使用Docker部署
# Dockerfile FROM python:3.9-slim WORKDIR /app COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt COPY . . CMD ["python", "auto_report.py"]
# docker-compose.yml
version: '3'
services:
reporter:
build: .
environment:
- API_URL=https://your-api.com/report
- TOKEN=your-token
- INTERVAL=300
restart: always
使用建议
-
选择方案：
- 简单场景：Shell + Cron
- 复杂场景：Python全功能版
- 高可靠性：带重试和熔断版本
-
安全考虑：
- 使用环境变量存储敏感信息
- 添加签名验证
- 使用HTTPS传输
-
监控和告警：
- 添加失败告警
- 记录日志
- 监控上报成功率
需要根据你的具体需求进行调整,比如API格式、认证方式、上报频率等。
