函数计算自动化运维实战1 -- 定时任务

本文涉及的产品
函数计算FC,每月15万CU 3个月
Serverless 应用引擎免费试用套餐包,4320000 CU,有效期3个月
简介: 阿里云函数计算是一个事件驱动的全托管计算服务。设置定时触发器,可以在某个时间点触发函数执行或者每隔5分钟触发函数一次;这样就可定时执行自定义的功能,比如这个功能就是运维脚本或者任务

函数计算

阿里云函数计算是一个事件驱动的全托管计算服务。通过函数计算,您无需管理服务器等基础设施,只需编写代码并上传。函数计算会为您准备好计算资源,以弹性、可靠的方式运行您的代码,并提供日志查询,性能监控,报警等功能。借助于函数计算,您可以快速构建任何类型的应用和服务,无需管理和运维。更棒的是,您只需要为代码实际运行消耗的资源付费,而代码未运行则不产生费用。

函数计算中的TimeTrigger

触发器是触发函数执行的方式。有时候您不想手动调用函数执行,您希望当某件事情发生时自动触发函数的执行,这个事情就是事件源。您可以通过配置触发器的方式设置事件源触发函数执行。

例如,设置定时触发器,可以在某个时间点触发函数执行或者每隔5分钟触发函数一次;函数计算timetrigger

专题传送门 => 函数计算进行自动化运维专题

定时任务自动化场景分析

定时任务示例场景1

某些账号ak需要定期更换,以确保ak安全;
在下面的代码示例中,授权service具有访问kms权限的能力,使用kms,先对一个具有创建和删除ak权限的ak加密密文解密,获取具有创建和删除ak权限的AK, 之后利用这个AK进行ak的创建和删除操作

说明: 除了使用kms加密解密来获取较大权限的AK, 通过函数计算环境变量的设置也是一种很好的方法

操作步骤

注:记得给函数的service的role设置访问kms权限

  • 给函数配置定时器,详情可参考定时触发函数
    image
  • 函数代码(函数计算已经内置了相关sdk,直接使用下面的代码即可)
# -*- coding: utf-8 -*-
import logging, time, json
from aliyunsdkcore import client
from aliyunsdkram.request.v20150501.CreateAccessKeyRequest import CreateAccessKeyRequest
from aliyunsdkram.request.v20150501.DeleteAccessKeyRequest import DeleteAccessKeyRequest
from aliyunsdkkms.request.v20160120.EncryptRequest import EncryptRequest
from aliyunsdkkms.request.v20160120.DecryptRequest import DecryptRequest
from aliyunsdkcore.auth.credentials import StsTokenCredential

# ak Encrypt content
AK_CiphertextBlob = "NmQyY2ZhODMtMTlhYS00MTNjLTlmZjAtZTQxYTFiYWVmMzZmM1B1NXhTZENCNXVWd1dhdTNMWVRvb3V6dU9QcVVlMXRBQUFBQUFBQUFBQ3gwZTkzeGhDdHVzMWhDUCtZeVVuMWlobzlCa3VxMlErOXFHWWdXXXHELLwL1NSZTFvUURYSW9lak5Hak1lMnF0R2I1TWUxMEJiYmkzVnBwZHlrWGYzc3kyK2tQbGlKb2lHQ3lrZUdieHN2eXZwSVYzN2Qyd1cydz09"

USER_NAME = "ls-test" # sub-account name
LOGGER = logging.getLogger()

def handler(event, context):
  creds = context.credentials
  sts_token_credential = StsTokenCredential(creds.access_key_id, creds.access_key_secret, creds.security_token)
  # this demo ecs and function in same region, if not in same region, you need change region_id to your ecs instance's region_id
  clt = client.AcsClient(region_id=context.region, credential=sts_token_credential)
  request = DecryptRequest()
  request.set_CiphertextBlob(AK_CiphertextBlob)
  response = _send_request(clt, request)
  ak_info = json.loads(response.get("Plaintext","{}"))
  if not ak_info:
    return "KMS Decrypt ERROR"
  ak_id = ak_info["ak_id"]
  ak_secret = ak_info["ak_secret"]
  LOGGER.info("Decrypt sucessfully with key id: {}".format(response.get("KeyId","{}")))

  clt2 = client.AcsClient(ak_id, ak_secret, context.region)
  request = CreateAccessKeyRequest()
  request.set_UserName(USER_NAME) # 给子账号ls-test创建AK
  response = _send_request(clt2, request)
  create_ak_id = response.get("AccessKey",{}).get("AccessKeyId")
  if not create_ak_id:
    return
  LOGGER.info("create ak {} sucess!".format(create_ak_id))
  
  time.sleep(10)
  
  request = DeleteAccessKeyRequest()
  request.set_UserName(USER_NAME)  
  request.set_UserAccessKeyId(create_ak_id)
  response = _send_request(clt2, request)
  LOGGER.info("delete ak {} sucess!".format(create_ak_id))
  
  return "OK"
  
# send open api request
def _send_request(clt, request):
    request.set_accept_format('json')
    try:
        response_str = clt.do_action_with_exception(request)
        LOGGER.debug(response_str)
        response_detail = json.loads(response_str)
        return response_detail
    except Exception as e:
        LOGGER.error(e)
AI 代码解读

AK 存在环境变量版本

# -*- coding: utf-8 -*-
import os, logging, time, json
from aliyunsdkcore import client
from aliyunsdkram.request.v20150501.CreateAccessKeyRequest import CreateAccessKeyRequest
from aliyunsdkram.request.v20150501.DeleteAccessKeyRequest import DeleteAccessKeyRequest

USER_NAME = "ls-test" # sub-account name
LOGGER = logging.getLogger()

def handler(event, context):
  ak_id = os.environ['AK_ID']
  ak_secret = os.environ['AK_SECRET']
  clt = client.AcsClient(ak_id, ak_secret, context.region)
  request = CreateAccessKeyRequest()
  request.set_UserName(USER_NAME) # 给子账号USER_NAME创建AK
  response = _send_request(clt, request)
  create_ak_id = response.get("AccessKey", "").get("AccessKeyId")
  if not create_ak_id:
    return
  LOGGER.info("create ak {} sucess!".format(create_ak_id))
  
  time.sleep(5)
  
  request = DeleteAccessKeyRequest()
  request.set_UserName(USER_NAME)  
  request.set_UserAccessKeyId(create_ak_id)
  response = _send_request(clt, request)
  LOGGER.info("delete ak {} sucess!".format(create_ak_id))
  
  return "OK"
  
# send open api request
def _send_request(clt, request):
    request.set_accept_format('json')
    try:
        response_str = clt.do_action_with_exception(request)
        LOGGER.info(response_str)
        response_detail = json.loads(response_str)
        return response_detail
    except Exception as e:
        LOGGER.error(e)
AI 代码解读

定时任务示例场景2

定期检查自己ecs对应暴露的端口,确保安全,比如你的ecs是一个网站服务器,可能只需要对外暴露80/443端口就行,如果出现0.0.0.0/0这种允许所有人访问其他端口,需要出现报警或者自动修复

操作步骤

注:记得给函数的service的role设置管理ecs权限

# -*- coding: utf-8 -*-
import logging
import json, random, string, time
from aliyunsdkcore import client
from aliyunsdkecs.request.v20140526.DescribeInstancesRequest import DescribeInstancesRequest
from aliyunsdkecs.request.v20140526.DescribeSecurityGroupAttributeRequest import DescribeSecurityGroupAttributeRequest
from aliyunsdkcore.auth.credentials import StsTokenCredential

LOGGER = logging.getLogger()
clt = None

# 需要检查的ecs列表, 修改成你的ecs id 列表
ECS_INST_IDS = ["i-uf6h07zdscdg9g55zkxx", "i-uf6bwkxfxh847a1e2xxx"]

def handler(event, context):
  creds = context.credentials
  global clt
  sts_token_credential = StsTokenCredential(creds.access_key_id, creds.access_key_secret, creds.security_token)
  # this demo ecs and function in same region, if not in same region, you need change region_id to your ecs instance's region_id
  clt = client.AcsClient(region_id=context.region, credential=sts_token_credential)
  invalid_perssions = {}
  for ecs_id in ECS_INST_IDS:
    ret = check_and_modify_security_rule(ecs_id)
    if ret:
      invalid_perssions[ecs_id] = ret
  return invalid_perssions

def check_and_modify_security_rule(instance_id):
  LOGGER.info("check_and_modify_security_rule, instance_id  is %s ", instance_id)
  request = DescribeInstancesRequest()
  request.set_InstanceIds(json.dumps([instance_id]))
  response = _send_request(request)
  SecurityGroupIds = []
  if response is not None:
    instance_list = response.get('Instances', {}).get('Instance')
    for item in instance_list:
      SecurityGroupIds = item.get('SecurityGroupIds', {}).get("SecurityGroupId", [])
      break
  if not SecurityGroupIds:
    LOGGER.error("ecs {} do not have SecurityGroupIds".format(instance_id))
    return 
  
  invalid_perssions = []
  
  for sg_id in SecurityGroupIds:
    request = DescribeSecurityGroupAttributeRequest()
    request.set_SecurityGroupId(sg_id)
    response = _send_request(request)
    LOGGER.info("Find a securityGroup id {}".format(sg_id))
    permissions = response.get("Permissions", {}).get("Permission",[])
    if not permissions:
      continue
    for permission in permissions:
        if permission["Direction"] == "ingress" and permission["SourceCidrIp"] == "0.0.0.0/0" \
              and permission["PortRange"] not in ["443/443", "80/80"]:
        LOGGER.error("ecs {0} , SecurityGroup id {1}, have a risk, need fix; permission = {2}".format(instance_id, sg_id, permission))
        invalid_perssions.append(permission)
        
  return invalid_perssions

# send open api request
def _send_request(request):
    request.set_accept_format('json')
    try:
        response_str = clt.do_action_with_exception(request)
        LOGGER.debug(response_str)
        response_detail = json.loads(response_str)
        return response_detail
    except Exception as e:
        LOGGER.error(e)
 
AI 代码解读
相关实践学习
【文生图】一键部署Stable Diffusion基于函数计算
本实验教你如何在函数计算FC上从零开始部署Stable Diffusion来进行AI绘画创作,开启AIGC盲盒。函数计算提供一定的免费额度供用户使用。本实验答疑钉钉群:29290019867
建立 Serverless 思维
本课程包括: Serverless 应用引擎的概念, 为开发者带来的实际价值, 以及让您了解常见的 Serverless 架构模式
rsong
+关注
目录
打赏
0
0
0
0
184
分享
相关文章
自动化运维的魔法:使用Python脚本简化日常任务
【8月更文挑战第50天】在数字化时代的浪潮中,自动化运维成为提升效率、减少人为错误的利器。本文将通过一个实际案例,展示如何利用Python脚本实现自动化部署和监控,从而让运维工作变得更加轻松和高效。我们将一起探索代码的力量,解锁自动化运维的神秘面纱,让你的工作环境焕然一新。
199 81
自动化运维:使用Python脚本简化日常任务
【10月更文挑战第36天】在数字化时代,运维工作的效率和准确性成为企业竞争力的关键。本文将介绍如何通过编写Python脚本来自动化日常的运维任务,不仅提高工作效率,还能降低人为错误的风险。从基础的文件操作到进阶的网络管理,我们将一步步展示Python在自动化运维中的应用,并分享实用的代码示例,帮助读者快速掌握自动化运维的核心技能。
181 3
自动化运维:编写基本的自动化脚本
自动化运维:编写基本的自动化脚本
121 1
自动化运维:使用Python脚本实现服务器批量管理
【8月更文挑战第2天】在现代IT运维领域,效率和准确性是衡量工作质量的关键指标。手动管理大量服务器不仅耗时且容易出错,因此自动化运维工具的开发变得至关重要。本文将介绍如何利用Python编写一个简单的自动化脚本,实现对多台服务器的批量管理,包括自动更新、配置同步以及日志收集等功能。通过实际案例展示,我们能够看到自动化运维如何提升工作效率并减少人为错误。
自动化运维:使用Python脚本实现批量部署
【8月更文挑战第2天】在现代IT基础设施管理中,自动化运维成为提升效率、减少人为错误的关键。本文将通过一个实际的Python脚本示例,展示如何实现服务器的批量部署,包括环境准备、代码实现及执行过程。文章旨在为运维工程师提供一种简化日常任务的方法,同时强调安全性和可维护性的重要性。
函数计算产品使用问题之如何实现项目自动化部署
函数计算产品作为一种事件驱动的全托管计算服务,让用户能够专注于业务逻辑的编写,而无需关心底层服务器的管理与运维。你可以有效地利用函数计算产品来支撑各类应用场景,从简单的数据处理到复杂的业务逻辑,实现快速、高效、低成本的云上部署与运维。以下是一些关于使用函数计算产品的合集和要点,帮助你更好地理解和应用这一服务。
函数计算产品使用问题之怎么集成自动化部署
函数计算产品作为一种事件驱动的全托管计算服务,让用户能够专注于业务逻辑的编写,而无需关心底层服务器的管理与运维。你可以有效地利用函数计算产品来支撑各类应用场景,从简单的数据处理到复杂的业务逻辑,实现快速、高效、低成本的云上部署与运维。以下是一些关于使用函数计算产品的合集和要点,帮助你更好地理解和应用这一服务。
|
10月前
|
自动化运维脚本集合
自动化运维脚本集合
109 0
自动化运维?看看Python怎样完成自动任务调度⛵
本文讲解如何使用Python的schedule库完成自动化调度程序,通过代『发送邮件』这一代码实例展示了构建任务与任务调度的全过程。
2768 4
自动化运维?看看Python怎样完成自动任务调度⛵
《函数计算最佳实践:基于函数计算实现 Serverless 自动化运维》电子版地址
函数计算最佳实践:基于函数计算实现 Serverless 自动化运维
117 0
《函数计算最佳实践:基于函数计算实现 Serverless 自动化运维》电子版地址

相关产品

  • 函数计算