클라우드 개발자 앙몬드

노드js에서나 보던 비동기 처리를, 파이썬에서도 보게 될 줄이야...

파일쓰기에서 비동기 처리가 필요하였기에 테스트성으로 개발한 용도입니다.

import boto3, json, pathlib, sys, asyncio, logging, itertools
from datetime import datetime
from aiofile import AIOFile, Reader, Writer, async_open

async def 내용을_생성하고_파일쓰기_호출하는_함수() :
	for i in range(10000) :
		한페이지 = []
		for j in range(10000) : 한페이지.append(str(i) + "\n")
		asyncio.create_task(파일쓰기함수(한페이지))
		#await asyncio.create_task(파일쓰기함수(한페이지))
	 
async def 파일쓰기함수(한페이지) :
	with open("temp2.txt", "a") as 파일 : 파일.writelines(한페이지)

async def 주황버섯의_함수() :
	print("시작!")
	시작시간 = datetime.now()
	await asyncio.create_task(내용을_생성하고_파일쓰기_호출하는_함수())
	끝난시간 = datetime.now()
	print(f"{시작시간} -> {끝난시간}")
	print(f"총 걸린시간 : {끝난시간 - 시작시간}")

def 핑크빈의_파일쓰기_검증함수() :
	print("검증몬 출동")
	검증결과 = { }
	with open("temp2.txt", "r") as 파일 :
		한줄 = 파일.readline()
		while 한줄 != "" :
			try : 
				검증결과[한줄] += 1
			except Exception as e :
				검증결과[한줄] = 1
			한줄 = 파일.readline()

	for k,v in 검증결과.items() : 
		if v != 10000 : print(f"{k} : {v}")
		#print(f"{k} : {v}")

if __name__ == "__main__" :

	#핑크빈의_파일쓰기_검증함수()
	#exit(0)

	물범의_이벤트루프 = asyncio.get_event_loop()
	물범의_이벤트루프.run_until_complete(주황버섯의_함수())
	물범의_이벤트루프.close()

import pandas as pd

d = [{'모듈': '주문API', '상태': 'Terminating'}, {'모듈': '전시프론트', '상태': 'Terminating'},{'모듈': '전시프론트', '상태': 'Terminating'}, {'모듈': '주문API', '상태': 'Terminating'}]

d_unique = pd.DataFrame(d).drop_duplicates().to_dict('records')

print(d_unique)
# pip3 install pandas  !!!!!

pandas 라이브러리를 이용하여, 리스트 엘리먼트가 딕셔너리 타입일 때, 중복을 제거하는 방법입니다.

만약 람다 레이어 및 디펜던시 의존성 계층이 무거워서, 라이브러리 추가가 힘들 경우, 하기와 같은 심플한 방법으로 처리하는 것도 좋은 방법!!

list_of_data = [{'모듈': '주문API', '상태': 'Terminating'}, {'모듈': '전시프론트', '상태': 'Terminating'},{'모듈': '전시프론트', '상태': 'Terminating'}, {'모듈': '주문API', '상태': 'Terminating'}]
list_of_data_uniq = []
for data in list_of_data:
    if data not in list_of_data_uniq:
        list_of_data_uniq.append(data)

쿠버네티스 노드에 대한 이슈가 자꾸 불거지기에, 계속 요즘따라 관제를 커스터마이징 하고 있습니댜

이전에 터졌던 EKS PLEG이슈에 이어 간혹 산발적으로 노드가 하나씩 다운되는 사태까지..

머리속으로는 이해하지만, 막상 시간이 부족하여 잘 만들지 못했던 관제를 이제야 꺼내보는군여.....

사실 고대의 존재들이 만들어두었던 방식을 제가 그대로 싹 흡수해서 만들게 된 케이스이기도 합니댜

아주 먼 옛날, 이때 만들면서 시간대역 때문에 시간 허비했던 그 관제시스템의 템플릿 버전이기도 하죠.

한편으론 감사합니다. 머나먼 고대의 존재들에게 이어받은 모든 지식을 토대로, 이렇게 제가 활용하고 있따는 것을 보면 훗훗..

(앙몬드가 가을타다보니 별 생각이 다 든댜)

devloper-angmond.tistory.com/6

상기 환경과는 다르게 지금의 환경은, 클라우드 리소스 접근이 어렵기 때문에, 젠킨스의 크론 빌드를 이용하여, 관제를 걸기로 하였습니댜

따라서, 준비되는 리소스는 다음과 같습니댜

0. 환경작업 패스

1. EC2 머신 (젠킨스 설치 / 크론탭으로 빌드 유발)

2. 크론탭 빌드 유발 시, 준비 되어야하는 파이썬3 스크립트

3. ES & 키바나 비주얼라이즈 with S3 Serverless Host

4. 노드가 비정상으로 감지되었을 시, 텔레그램으로 관제 발송

간단한 아키텍처는 이와 같습니댜

텔레그램에 대한 관제는 정말 별거 없지만, 가이드가 없으면 갈피잡기가 너무 어렵기 떄문에, 훗날 서브로 다시 작성해둘게요.

"""
텔레그램 메세지 전송 헬퍼

@author 클라우드 개발자 앙몬드
@since 2020-11-02

[2020-11-02] 최초생성
"""

import telegram

토큰 = '봇에게 접근하기 위한 토큰값입니다. 숫자:문자 로 이루어져있어요'
#텔레그램_채팅방_아이디 = "채팅방마다 시퀀스가 존재합니다"  #내 개인방
텔레그램_채팅방_아이디 = "채팅방마다 시퀀스가 존재합니다"   #데브옵스 엔지니어 방 
봇 = telegram.Bot(token=토큰)

#채팅방 아이디를 얻어내기 위한, json값 획득
#for i in 봇.getUpdates():
#    print(i.message)

def 텔레그램_메세지보내기(메세지) :
    봇.sendMessage(chat_id=텔레그램_채팅방_아이디, text=메세지) 

# -*- coding: utf-8 -*- 
import json
import time
import os
import decimal
import sys
import subprocess
import yaml
import boto3 
from requests_aws4auth import AWS4Auth
from elasticsearch import Elasticsearch, RequestsHttpConnection
from datetime import datetime, timedelta
from os import rename, listdir
from angmond_es_helper import *
import telegram_helper

"""
쿠버네티스 클러스터 노드 상태 체크의 백엔드 로직

@since 2020-10-29
@author 클라우드 개발자 앙몬드

[2020-10-29] 최초생성
"""

비정상노드리스트 = []
ES입력리스트 = []

메세지내용 = ""
노드명 = ""
노드상태 = ""
노드_실행기간 = ""
노드그룹 = ""
today = datetime.today()

ES_노드상태_인덱스 = "devops-eks-node-status-"
ES_노드사용_인덱스 = "devops-eks-node-usage-"
ES_노드그룹사용_인덱스 = "devops-eks-nodegroup-usage-"

ES커넥터 = get_aws_es_connector(get_aws_es_credentials(True), "ES_호스트주소!")
ES타임스탬프 = (today - timedelta(hours=9)).isoformat()

if __name__ == "__main__" : 
    비정제_전체_노드_요약리스트 = subprocess.check_output("kubectl get node -L group | grep -v NAME", shell=True, universal_newlines=True).strip()
    노드리스트 = str(비정제_전체_노드_요약리스트).split("\n")

    노드_FO_CPU = []	#프론트 모듈중 CPU에 특화된 노드들이 구성된 노드그룹
    노드_FO_MEM = []	#프론트 모듈 중 메모리에 특화된 노드들이 구성된 노드그룹
    노드_BO_MEM = []	#백오피스/현업/ERP 모듈 중 메모리에 특화된 노드들이 구성된 노드그룹

    # (0) 비정상노드 탐지를 위한 정제부분
    완전정제된_노드리스트 = list(map(lambda x : (" ".join(x.split())).split(" "), 노드리스트))

    노드_FO_CPU = list(filter(lambda x : x[5] == "fo_cpu", 완전정제된_노드리스트))
    노드_FO_MEM = list(filter(lambda x : x[5] == "fo_mem", 완전정제된_노드리스트))
    노드_BO_MEM = list(filter(lambda x : x[5] == "bo_mem", 완전정제된_노드리스트))

    # (1) 비정상노드 상태값 판별 & 텔레그램 관제
    for 노드 in 완전정제된_노드리스트 :
        """
        노드정보는 리스트로..
        0 : NAME
        1 : STATUS
        2 : Role
        3 : AGE
        4 : EKS버전
        5 : 노드 그룹
        """  
        
        메모리부하_상태 = subprocess.check_output("kubectl describe node " + 노드[0] + " | grep '  MemoryPressure'", shell=True, universal_newlines=True).strip()
        볼륨부하_상태 = subprocess.check_output("kubectl describe node " + 노드[0] + " | grep '  DiskPressure'", shell=True, universal_newlines=True).strip()
        PID부하_상태 = subprocess.check_output("kubectl describe node " + 노드[0] + " | grep '  PIDPressure'", shell=True, universal_newlines=True).strip()
        쿠블릿_상태 = subprocess.check_output("kubectl describe node " + 노드[0] + " | grep '  Ready'", shell=True, universal_newlines=True).strip()
        
        관제메세지 = ""
        if "True" in 메모리부하_상태 :
            관제메세지 = 관제메세지 + "MemoryPressure 상태 감지 "
            메모리부하_상태 = True
        else : 메모리부하_상태 = False

        if "True" in 볼륨부하_상태 :
            관제메세지 = 관제메세지 + "DiskPressure 상태 감지 " 
            볼륨부하_상태 = True
        else : 볼륨부하_상태 = False

        if "True" in PID부하_상태 :
            관제메세지 = 관제메세지 + "PIDPressure 상태 감지 " 
            PID부하_상태 = True
        else : PID부하_상태 = False

        if "False" in 쿠블릿_상태 :
            관제메세지 = 관제메세지 + "쿠블릿 비정상 감지 " 
            쿠블릿_상태 = False
        else : 쿠블릿_상태 = True

        # 상태 이상 발견 시, 텔레그램 관제 문자 발송
        if len(관제메세지) != 0 :
            telegram_helper.텔레그램_메세지보내기(노드[0] + " -> " + 관제메세지)
        
        # ES인덱싱
        es_다큐먼트 = {
            "node" : 노드[0],
            "MemoryPressure" : 메모리부하_상태,
            "DiskPressure" : 볼륨부하_상태,
            "PIDPressure" : PID부하_상태,
            "KubeletReady" : 쿠블릿_상태,
            "@timestamp" : ES타임스탬프,
        }
        print(es_다큐먼트)
        ES인덱스 = ES_노드상태_인덱스 + str(today.year) + '.' + str(today.month)
        create_es_index(ES커넥터, ES인덱스, es_다큐먼트)       

    # (2) 그룹별 CPU/Mem 사용률 
    비정제_노드자원사용률리스트 = subprocess.check_output("kubectl top node | grep -v NAME", shell=True, universal_newlines=True).strip()
    노드자원사용률리스트 = str(비정제_노드자원사용률리스트).split("\n")

    FO_CPU_노드만 = list(map(lambda y : y[0], list(filter(lambda x : x[5] == "fo_cpu", 완전정제된_노드리스트))))
    FO_MEM_노드만 = list(map(lambda y : y[0], list(filter(lambda x : x[5] == "fo_mem", 완전정제된_노드리스트))))
    BO_MEM_노드만 = list(map(lambda y : y[0], list(filter(lambda x : x[5] == "bo_mem", 완전정제된_노드리스트))))
    FO_CPU_리소스필터링_리스트 = []
    FO_MEM_리소스필터링_리스트 = []
    BO_MEM_리소스필터링_리스트 = []
    for 노드리소스사용률 in 노드자원사용률리스트:
        정제된_노드리소스_사용정보 = (" ".join(노드리소스사용률.split())).split(" ")        
        """
        [0] 노드명
        [1] CPU 사용량
        [2] CPU 사용률
        [3] Mem 사용량
        [4] Mem 사용률
        """

        es_다큐먼트 = {
            "node" : 정제된_노드리소스_사용정보[0],
            "cpu_usage" : int(정제된_노드리소스_사용정보[1].replace("m", "")),
            "cpu_use_rate" : int(정제된_노드리소스_사용정보[2].replace("%", "")),
            "mem_usage" : int(정제된_노드리소스_사용정보[3].replace("Mi", "")),
            "mem_use_rate" : int(정제된_노드리소스_사용정보[4].replace("%", "")),
            "@timestamp" : ES타임스탬프,
        }
        ES인덱스 = ES_노드사용_인덱스 + str(today.year) + '.' + str(today.month)
        print(es_다큐먼트)
        create_es_index(ES커넥터, ES인덱스, es_다큐먼트)

        if 정제된_노드리소스_사용정보[0] in FO_CPU_노드만 :
            es_다큐먼트["node_group"] = "fo_cpu"
            FO_CPU_리소스필터링_리스트.append(es_다큐먼트)
        elif 정제된_노드리소스_사용정보[0] in FO_MEM_노드만 :
            es_다큐먼트["node_group"] = "fo_mem"
            FO_MEM_리소스필터링_리스트.append(es_다큐먼트) 
        elif 정제된_노드리소스_사용정보[0] in BO_MEM_노드만 :
            es_다큐먼트["node_group"] = "bo_mem"
            BO_MEM_리소스필터링_리스트.append(es_다큐먼트) 

    for FO_CPU_정보 in FO_CPU_리소스필터링_리스트:
        ES인덱스 = ES_노드그룹사용_인덱스 + str(today.year) + '.' + str(today.month)
        create_es_index(ES커넥터, ES인덱스, FO_CPU_정보)
    for FO_MEM_정보 in FO_MEM_리소스필터링_리스트:
        ES인덱스 = ES_노드그룹사용_인덱스 + str(today.year) + '.' + str(today.month)
        create_es_index(ES커넥터, ES인덱스, FO_MEM_정보)
    for BO_MEM_정보 in BO_MEM_리소스필터링_리스트:
        ES인덱스 = ES_노드그룹사용_인덱스 + str(today.year) + '.' + str(today.month)
        create_es_index(ES커넥터, ES인덱스, BO_MEM_정보) 

AWS EKS를 쓰다 도저히 잡히지 않는 에러가 발생하여 한창 고민한 하루입니다.

상황에 대한 시나리오입니다.

(1) 쿠버네티스 클러스터 노드 중 일부 노드 그룹 (레이블로 그룹핑) 의 노드가 비정상 동작을 감지

(2) 비정상 동작 시, 해당 노드의 파드들이 정상동작되지 않는 사태

    -> OOM (unable to create new native Thread) , Connection Exception, 뭐 등등등등.. 일반적으로 스레드 풀 획득 실패

(3) 파드가 다운되고, HPA 스케일러가 이를 복구하며, 무한반복 및 파드 스케줄에 펜딩 걸리는 난장판이 펼쳐지기 시작 (와우!)

아직도 쿠버네티스 운전 잘 못하는 앙몬드의 트러블슈팅 시작..

(1) 2번 상황에서, 모든 이슈가 하나의 공통점으로 수렴하는데, 네트워크 자원에 대한 문제가 있었댜..? 

=> 문제는 해당 노드의 모든 파드가 그런 증상을 보이진 않았으며, 만약 그렇다면 ENI 매니지드 서비스에 대한 클라우드 AZ단위 장애일 가능성이 높고, 그에 대한 알람을 받은 적이 없으므로 패스

(2) 파드 내부 컨테이너 런타임인 어플리케이션의 문제댜..?

=> 그럼 다른 노드의 같은 어플리케이션들은 왜 잘 돌아가는지에 대한 확신이 없으므로, 이 또한 아니댜..

(3) 노드가 터졌댜..?

=> 반드시 그런건 아니고, 노드에 할당된 (Allocation) 은 노드 내부 프로세스 리소스 사용률을 고려하여, 70%까지만 유지하였고, 또한 이 당시에 파드가 제한값까지 치는 상황은 일어나지 않았댜.. 그런데.. NodeNotReady는 왜 뜨지.. 안의 파드들은 정상 기동중이댜..

(4) 뭐냐 그럼...?

사실 AWS EKS를 살펴보다 보면, 생각보다 간과하고서 넘어간 부분이 있습니다.

kubectl describe node {노드명}

커맨드로 쿠버 API를 호출하면, 맨날 지나쳤던 로그의 일부분. (맨날 정상이었으니 그냥 넘어갔지....)

노드의 상태값인데, 자세히 보면, Ready Type이 False. 즉 노드가 NotReady 상태로 나타났고, 이 노드를 매니징해야하는 Kubelet, 쿠버 워커노드의 중간관리자가 일할 준비가 안되어있죠.

그리고, 그 옆에 왜 쿠블릿이 정상동작하지 못한 사유가 나옵니다

"PLEG is not healthy: pleg was last seen active .... ~~"

당신 뭡니까..? PLEG요..? 아직 쿠버 초보운전이라지만, 이건 또 뭡니까..? 

developers.redhat.com/blog/2019/11/13/pod-lifecycle-event-generator-understanding-the-pleg-is-not-healthy-issue-in-kubernetes/

레드햇 공홈의 영어 원문입니다. 워커노드의 중간관리자인 쿠블릿 안에 위치한 모듈이고, 컨테이너 런타임의 상태를 조절하는 녀석이라고 설명하고 있습니다. 자세한 사항은 상기 레퍼런스를 참조해주시면 좋을 것 같아요! (머리속에 집어넣을 것도 많은데 CKA도 따야 하는건가..)

자 그럼 여기서 문제를 던져보면, PLEG모듈이 왜 컨테이너들의 상태를 비정상이라고 판단하였는지, 그리고 쿠블릿이 정상적으로 동작하던 파드들을 마구마구 죽이면서 저 에러가 난 것일까라는 의문에 도달하고 마는데!

* 그렇다면 PLEG에 대한 로그는 어떻게 볼까?

보통은, 노드의 이벤트 필드에 그 값이 나타나지만, 많은 로그를 보여주진 않습니댜.

따라서, 이에 대한 로그를 시간대별로 볼 방법이 필요한데....

* journalctl -xfu kubelet (OS별로 커맨드는 다를 수 있어요! 제가 사용하고 있는 것은 AML2, 레드햇계열 리눅스입니다)

워커노드의 쿠블릿에 대한 로그를 보여줍니다.

그러나, 이 로그와, 실제 어플리케이션의 활동로그를 보면 상이한 차이가 발생했는데..

    - kubelet : pleg가 그러던데, 너님 이상하니까 아웃

    - pod : ?????? 아니 왜 선량한 저를...

지금 이 상황입니다 정말..

결국, 왜 쿠블릿의 pleg 모듈이 저 파드들을 전부 이상한 사람 취급해서, 못살게 구나는 야간 철야로도 밝혀지지 않게 되었고, 다음 날 EKS 실제 테크니션의 답변을 듣게 됩니다.

아.. 지금 야간 작업 해야하니 간단하게만 결론 내고 가죠..

EKS 1.16 미만 의 버전에서, 노드의 CPU가 100% 가까이 사용될 때, pleg가 오작동을 하는 버그가 존재합니다.

글로벌적으로도 많이 문의가 들어온 내역입니다.

EKS의 버그이며, 현재 사용하고 있는 1.14 버전을 업그레이드 해주시면 됩니다.

.....?

그리고 이건 쿠블릿 아키텍처입니다.

programmer.help/images/blog/30f7ca203efd66487c602c86184b5b87.jpg

앙몬드입니댜

이번에는 엘라스틱서치와 관련한 사항입니댜

현재 저희 환경의 경우, cost optimize를 위한 방안을 여러 각도로 시도해보고 있는데,

제 관할에 있는 AWS 엘라스틱서치 노드 클러스터도 비용 절약 측면에 포함되어 있습니댜

현재 상황은 다음과 같습니다

 (1) 실시간으로 엘라스틱서치에서 검색 가능한 다큐먼트는 2주치만 유지하고, 이외에는 삭제한다.

 (2) 단, 이에 대한 스냅샷은 유지하되, 스냅샷은 4주만 보관하고 그 이전의 날짜는 삭제한다.

다음과 같은 라이프사이클링을 실행하기 위한 람다가 배치성으로 실행되고 있죠

단, 개발팀 측 문의로는 과거 이슈에 대하여 디버깅할 필요가 있어, 이에 대한 로그를 복구해줄 것을 요구했는데요.

그에 따라 스냅샷으로 들어간 엘라스틱서치 데이터들을 복원하는 람다를 한번 개발해보았습니다.

물론 가장 쉬운 것은, 모든 스냅샷을 통틀어 복구하는 것이 가장 단순하지만, 어쨌든 회사 정책에 따라 비용 최적화란 흐름을 무시할 순 없다보니, 특정 스냅샷의 특정 인덱스만 복구해보는 람다를 개발하는 방향으로 가게 되었습니댜 :)

AWS 엘라스틱서치 복구의 경우, AWS 화이트페이퍼는 curl / http 로 호출하는 request 모듈을 사용하는 방법으로 가이드가 되어 있으니, 이 방법이 편하시다면 가이드를 따라가시는 것도 한 방법입니다 :)

docs.aws.amazon.com/ko_kr/elasticsearch-service/latest/developerguide/es-managedomains-snapshots.html

자 이제 다시 돌아와서.. 앙몬드 쪽의 방향으로 돌려보면 람다 코드가 있는데요...

단, 이 람다를 돌리기 위한 조건!!!

 (1) requests_aws4auth 라이브러리와 elasticsearch 라이브러리가 람다 레이어에 포함되어 있어야 합니댜!!!!

 (2) 위에 꺼 딱 하나 입니댜 (아 그리고 IAM과 네트워크 환경설정은 당연히 되어있겠죠? 훗) 

환경변수로 입력값을 제어하면 됩니댜 :)

import json, boto3, os
from requests_aws4auth import AWS4Auth
from elasticsearch import Elasticsearch, RequestsHttpConnection

"""
AWS 엘라스틱서치 스냅샷 복구 람다

@since 2020-10-26
@author 클라우드 개발자 앙몬드

[2020-10-26] 최초생성 (클라우드 개발자 앙몬드)
"""

#### 환경변수 선언부
실행옵션 = os.environ['OPTION_FOR_EXECUTION']
스냅샷명 = os.environ["QUERY_SNAPSHOT_NAME"]
복구할_인덱스명 = os.environ["INDEX_TO_RESTORE"]
엘라스틱서치_엔드포인트 = os.environ["ES_ENDPOINT"]
레파지토리 = os.environ["ES_SNAPSHOT_REPOSITORY"]

호스트 = 엘라스틱서치_엔드포인트.replace("https://", "")
토큰 = boto3.Session().get_credentials()
aws_인증문서 = AWS4Auth(토큰.access_key, 토큰.secret_key, "ap-northeast-2", "es", session_token=토큰.token)
es_커넥터 = Elasticsearch(hosts = [{'host': 호스트, 'port': 443}],http_auth = aws_인증문서,use_ssl = True,verify_certs = True,connection_class = RequestsHttpConnection,timeout = 300)

def lambda_handler(event, context):
    
    """ 
        (0) 환경변수 설정 및 확인 
            [실행옵션]
                query_snapshot : 스냅샷 전체 조회
                query_snapshot_indices : 특정 스냅샷에 담긴 인덱스 전체조회 
                restore : 인덱스 복구
                default는 query_snapshot 모드입니다. 
            [스냅샷명]
                query_snapshot_indices 모드와 restore 모드에서 사용됩니다.
                2개 모드에서, 스냅샷명이 없을 경우 오류를 리턴합니다.
            [복구할_인덱스명]
                복구할 인덱스는 단건으로만 가능하게 제한한다.
                스냅샷명 / 인덱스명이 미리 지정되어 있지 않다면 오류를 리턴한다
    """
    print(f"{'*'*70}")
    print(f"입력된 실행옵션 : {실행옵션}")
    print(f"입력된 인덱스 : {복구할_인덱스명}")
    print(f"{'*'*70}")
    
    """
        (1) ES 스냅샷 조회 및 복구 처리 람다 코어
    """
    if 실행옵션 == "query_snapshot_indices" : # 스냅샷의 인덱스 조회모드
        print("\n\n\n")
        print("[query_snapshot_indices] 모드 - 특정 스냅샷의 인덱스 조회") 
        리스트_스냅샷 = es_커넥터.snapshot.get(repository=레파지토리, snapshot="_all")["snapshots"]
        필터링된_스냅샷 = list(filter(lambda x : x["snapshot"] == 스냅샷명, 리스트_스냅샷))[0]
        print(f"{'*'*70}")
        print(*필터링된_스냅샷['indices'], sep='\n')
        #print([ES인덱스 for ES인덱스 in 필터링된_스냅샷['indices']])
        print(f"{'*'*70}")
        
        
    elif 실행옵션 == "restore" : # 스냅샷 복구모드
        print("\n\n\n")
        print("[query_snapshot_indices] 모드 - 특정 스냅샷의 인덱스 조회") 
        결과 = es_커넥터.snapshot.restore(
            레파지토리, 스냅샷명, 
            body={
                "indices": 복구할_인덱스명,
                "include_global_state": False   # 얜 뭐지
            }
        ) 
        print(f"{'*'*70}")
        print(결과)
        
        
    else : # 복구가능 스냅샷 전체 조회모드
        print("\n\n\n")
        print("[query_snapshot / default] 모드 - 스냅샷 전체 조회")
        리스트_스냅샷 = es_커넥터.snapshot.get(repository=레파지토리, snapshot="_all")["snapshots"]
        print(f"{'*'*70}")
        for dict_스냅샷아이템 in 리스트_스냅샷 :
            print(dict_스냅샷아이템["snapshot"])
        print(f"{'*'*70}") 
    
    # TODO implement
    return {
        'statusCode': 200,
        'body': json.dumps('Success!!!')
    }

import os
import boto3
import json
import decimal
from boto3.dynamodb.conditions import Key, Attr
from botocore.exceptions import ClientError

"""
AWS DynamoDB 헬퍼 모듈
@desc
    파이썬 런타임의 람다가, DynamoDB에 쿼리를 할 수 있게 해주는 헬퍼 모듈
@since 2020-10-22
@author angmond
"""

# DynamoDB 아이템 조회
def get_item(dynamodb_table_name, **key_info):
    ddb = boto3.resource('dynamodb')
    table = ddb.Table(dynamodb_table_name) #DynamoDB 테이블명

    try:
        response = table.get_item(
            Key={
                'PK': key_info["DDB_파티션키!"],
                'SK': key_info["DDB_소트키!"]
            }
        )
    except ClientError as e:
        print(e.response['Error']['Message'])
        return e
    else:
        item = response['Item']
        print("GetItem succeeded:")
        print(json.dumps(item, indent=4, cls=DecimalEncoder))
        return item

# DynamoDB 아이템 생성
def put_item(dynamodb_table_name, ddb_item):

    ddb = boto3.resource('dynamodb')
    table = ddb.Table(dynamodb_table_name) #DynamoDB 테이블명

    response = table.put_item(
        Item = {
            'PK': ddb_item.파티션키,
            'SK': ddb_item.소트키,
            'data1' : ddb_item.데이터1,
            'data2' : ddb_item.데이터2,
        }
    )

    return response

# DynamoDB 아이템 조회
def get_item(dynamodb_table_name, **key_info) :
    pass

# DynamoDB 아이템 삭제
def delete_item(dynamodb_table_name, **key_info):
    pass

# Helper class to convert a DynamoDB item to JSON.
class DecimalEncoder(json.JSONEncoder):
    def default(self, o):
        if isinstance(o, decimal.Decimal):
            if o % 1 > 0:
                return float(o)
            else:
                return int(o)
        return super(DecimalEncoder, self).default(o)

AWS DynamoDB의 트랜잭션 제어를 위해 간단히 만들었던 파이썬3 DDB헬퍼 입니다.

아직 삭제 등 더 구현해야 할 헬퍼 함수가 많지만, 

boto3를 이용해, DynamoDB 인터페이스 커넥터를 생성하고, 파티션키/소트키/기타 데이터 등을 조합해 트랜잭션을 진행하는 모듈입니다.

이렇게 보니, 진짜 라이브러리 만드는 사람들 참 대단한거같네요.. 얼마나 수많은 케이스가 있는데, 이런 것을 고려해서 함수나 모듈을 만드는건지... (앙몬드는 글쓰다가도 감탄한댜)

아직은 1편 초짜라서 ,하나하나 추가될 예정입니다 :)