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

本文涉及的产品
Serverless 应用引擎 SAE,800核*时 1600GiB*时
服务治理 MSE Sentinel/OpenSergo,Agent数量 不受限
性能测试 PTS,5000VUM额度
简介: 阿里云函数计算是一个事件驱动的全托管计算服务。通过函数计算,您无需管理服务器等基础设施,只需编写代码并上传。函数计算会为您准备好计算资源,以弹性、可靠的方式运行您的代码,并提供日志查询,性能监控,报警等功能。借助于函数计算,您可以快速构建任何类型的应用和服务,无需管理和运维。更棒的是,您只需要为代码实际运行消耗的资源付费,而代码未运行则不产生费用。

函数计算

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

函数计算中的TimeTrigger

触发器是触发函数执行的方式。有时候您不想手动调用函数执行,您希望当某件事情发生时自动触发函数的执行,这个事情就是事件源。您可以通过配置触发器的方式设置事件源触发函数执行。
例如,设置定时触发器,可以在某个时间点触发函数执行或者每隔5分钟触发函数一次;函数计算timetrigger

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

定时任务自动化场景分析

定时任务示例场景1

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

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

操作步骤

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

_1

  • 函数代码(函数计算已经内置了相关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)

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)

定时任务示例场景2

定期检查自己ecs对应暴露的端口,确保安全,比如你的ecs是一个网站服务器,可能只需要对外暴露80端口就行,如果出现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":
        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)

“ 阿里巴巴云原生微信公众号(ID:Alicloudnative)关注微服务、Serverless、容器、Service Mesh等技术领域、聚焦云原生流行技术趋势、云原生大规模的落地实践,做最懂云原生开发者的技术公众号。”

相关实践学习
【文生图】一键部署Stable Diffusion基于函数计算
本实验教你如何在函数计算FC上从零开始部署Stable Diffusion来进行AI绘画创作,开启AIGC盲盒。函数计算提供一定的免费额度供用户使用。本实验答疑钉钉群:29290019867
建立 Serverless 思维
本课程包括: Serverless 应用引擎的概念, 为开发者带来的实际价值, 以及让您了解常见的 Serverless 架构模式
相关文章
|
9天前
|
运维 安全 应用服务中间件
自动化运维的利剑:Ansible实战应用
【9月更文挑战第24天】在现代IT基础设施的快速迭代与扩展中,自动化运维成为提升效率、保障稳定性的关键。本文将深入探讨Ansible这一流行的自动化工具,通过实际案例分析其如何简化日常运维任务,优化工作流程,并提高系统的可靠性和安全性。我们将从Ansible的基础概念入手,逐步深入到高级应用技巧,旨在为读者提供一套完整的Ansible应用解决方案。
|
27天前
|
运维 Ubuntu Devops
自动化运维工具的魅力:Ansible入门
【9月更文挑战第5天】在快速变化的IT世界里,自动化运维不再是可选项,而是必需品。Ansible,一款简单却强大的自动化工具,正成为众多DevOps工程师的首选。本文将带你了解Ansible的基本概念、安装步骤以及如何编写简单的Playbook,从而开启你的自动化之旅。
67 36
|
4天前
|
机器学习/深度学习 运维 Cloud Native
构建高效运维体系:从自动化到智能化的演进之路
在当今数字化时代,运维作为信息技术的重要支柱,其效率与创新能力直接关系到企业信息系统的稳定性和业务连续性。本文将探讨如何通过技术手段,实现运维从传统手工操作向自动化、智能化的转变,进而构建一个高效、可靠的运维体系。我们将从自动化工具的应用开始,逐步深入到智能运维的实践,最终展望云原生架构下的运维未来趋势。
|
7天前
|
机器学习/深度学习 人工智能 运维
构建高效运维体系:从自动化到智能化的演进之路
在当今数字化时代,运维作为保障企业IT系统稳定运行的关键环节,正经历着前所未有的变革。本文将探讨如何通过实施自动化和引入智能化技术,构建一个更加高效、可靠的运维体系,以应对日益复杂的业务需求和技术挑战。
19 1
|
24天前
|
存储 弹性计算 运维
自动化监控和响应ECS系统事件
阿里云提供的ECS系统事件用于记录云资源信息,如实例启停、到期通知等。为实现自动化运维,如故障处理与动态调度,可使用云助手插件`ecs-tool-event`。该插件定时获取并转化ECS事件为日志存储,便于监控与响应,无需额外开发,适用于大规模集群管理。详情及示例可见链接文档。
|
18天前
|
机器学习/深度学习 运维 安全
构建高效运维体系:从自动化到智能化的演进之路
在数字化转型的浪潮中,运维管理作为信息技术基础设施的重要支柱,正经历着从传统手工操作向自动化、智能化的深刻变革。本文将探讨如何通过引入自动化工具和平台,实现运维流程的标准化与效率提升;进而利用大数据分析和人工智能技术,迈向预测性维护和智能决策支持的高级阶段。通过案例分析,揭示成功转型的关键因素,为运维专业人士提供一套可借鉴的升级路径。
|
15天前
|
机器学习/深度学习 人工智能 运维
构建高效运维体系:从自动化到智能化的演进之旅
在当今数字化时代,运维作为信息技术领域的核心组成部分,其重要性日益凸显。随着企业业务的不断扩展和技术的日新月异,传统手工运维方式已难以满足现代IT架构的需求。因此,构建一个高效、智能的运维体系成为业界共识。本文将探讨如何通过自动化和智能化手段,实现运维效率的质的飞跃,并分享一些成功案例与实践经验。
|
20天前
|
机器学习/深度学习 运维 监控
构建高效运维体系:从自动化到智能化的演进
在当今信息技术飞速发展的时代,运维领域正经历着从传统手动操作向高度自动化和智能化转变。本文旨在探讨如何通过集成自动化工具、实施持续集成与持续部署(CI/CD)、采用容器化技术和探索AIOps实践,构建一个高效、可靠的运维体系。我们将分析这些技术如何相互配合,提升运维效率,降低系统故障率,并实现快速响应市场变化的能力。通过案例分析和最佳实践的分享,本文将为IT专业人员提供一套实用的指南,帮助他们在数字化转型的浪潮中,保持运维工作的前瞻性和竞争力。
|
21天前
|
机器学习/深度学习 数据采集 运维
智能化运维:打造高效、自动化的IT系统
在数字化转型的浪潮中,企业对于IT系统的依赖程度日益加深。如何确保系统的高效运行和快速响应,成为摆在每一个IT管理者面前的难题。本文将探讨智能化运维的概念、实施步骤及其带来的变革,旨在为读者提供一套构建自动化、智能化IT运维体系的思路和方法。
|
21天前
|
机器学习/深度学习 数据采集 人工智能
智能化运维的探索之旅:从自动化到人工智能
在数字化浪潮中,运维领域正经历一场革命。本文将带你领略从传统手动操作到自动化脚本,再到集成人工智能的智能运维平台的演变之路。我们将探讨如何通过技术创新提升效率、降低成本并增强系统的可靠性和安全性。文章不仅分享技术演进的故事,还提供了实现智能化运维的实践策略和未来趋势的展望。

相关产品

  • 函数计算
  • 下一篇
    无影云桌面