[Python] Exportieren Sie regelmäßig mit Lambda aus CloudWatch-Protokollen nach S3

Exportieren Sie die in CloudWatch-Protokollen gespeicherten Protokolle regelmäßig in S3, das in [AWS] CloudFormation erstellt wurde, um einen S3-Bucket zu erstellen und Lebenszyklusregeln festzulegen. Dies ist die Geschichte der Entwicklung von Lambda in Python.

Ich habe ein GitHub-Repository für einfache Testversionen erstellt-> homoluctus / lambda-cwlogs-s3

Bedarf

--Exportieren Sie das Protokoll des Vortages jeden Tag nach JST 14:00

Entwicklung von

Sprache

Entwicklungsbibliothek

Produktionsbibliothek

Bereitstellungstool

AWS-Service

Code

Es ist unmöglich, den gesamten Code einzugeben, daher wird nur der Hauptcode zur Erklärung angegeben Den vollständigen Quellcode finden Sie unter Repositories.

Klasse für die zu exportierende Protokollgruppeneinstellung

Aktiviert, um mehrere zu exportierende Protokollgruppen wie TypeScript-Schnittstelle festzulegen. Wenn Sie eine zu exportierende Protokollgruppe hinzufügen möchten, erben Sie einfach die LogGroup-Klasse.

Der zu exportierende S3-Objektschlüssel ist leicht verständlich, in welchem Protokollgruppenprotokoll. Wie ich in der Dokumentzeichenfolge geschrieben habe, hat es eine hierarchische Struktur wie dest_bucket / dest_obj_first_prefix oder log_group / dest_obj_final_prefix / *. Wenn dest_obj_first_prefix nicht angegeben ist, wird der Name log_group eingegeben. Nach "*" sieht es aus wie "Task-ID / Protokoll-Stream / Datei exportieren". Dies wird automatisch hinzugefügt und kann nicht gesteuert werden.

class LogGroup(object, metaclass=ABCMeta):
    """Konfigurationsbasisklasse zum Exportieren von CloudWatch-Protokollen nach S3

So fügen Sie eine zu exportierende Protokollgruppe hinzu

    class Example(LogGroup):
        log_group = 'test'
    """

    # log_Gruppe ist erforderlich, sonst optional
    log_group: ClassVar[str]

    log_stream: ClassVar[str] = ''
    start_time: ClassVar[int] = get_specific_time_on_yesterday(
        hour=0, minute=0, second=0)
    end_time: ClassVar[int] = get_specific_time_on_yesterday(
        hour=23, minute=59, second=59)
    dest_bucket: ClassVar[str] = 'lambda-cwlogs-s3'
    dest_obj_first_prefix: ClassVar[str] = ''
    dest_obj_final_prefix: ClassVar[str] = get_yesterday('%Y-%m-%d')

    @classmethod
    def get_dest_obj_prefix(cls) -> str:
        """Holen Sie sich das vollständige S3-Objektpräfix

Hierarchische Struktur von S3
        dest_bucket/dest_obj_first_prefix/dest_obj_final_prefix/*

        Returns:
            str
        """

        first_prefix = cls.dest_obj_first_prefix or cls.log_group
        return f'{first_prefix}/{cls.dest_obj_final_prefix}'

    @classmethod
    def to_args(cls) -> Dict[str, Union[str, int]]:
        args: Dict[str, Union[str, int]] = {
            'logGroupName': cls.log_group,
            'fromTime': cls.start_time,
            'to': cls.end_time,
            'destination': cls.dest_bucket,
            'destinationPrefix': cls.get_dest_obj_prefix()
        }

        if cls.log_stream:
            args['logStreamNamePrefix'] = cls.log_stream

        return args

CloudWatch Logs-Client

Die folgenden beiden werden in der CloudWatch-Protokoll-API verwendet

@dataclass
class Exporter:
    region: InitVar[str]
    client: CloudWatchLogsClient = field(init=False)

    def __post_init__(self, region: str):
        self.client = boto3.client('logs', region_name=region)

    def export(self, target: Type[LogGroup]) -> str:
        """Exportieren Sie eine beliebige CloudWatch-Protokollprotokollgruppe nach S3

        Args:
            target (Type[LogGroup])

        Raises:
            ExportToS3Error

        Returns:
            str:TaskId in Antwort von CloudWatch Logs API enthalten
        """

        try:
            response = self.client.create_export_task(
                **target.to_args())  # type: ignore
            return response['taskId']
        except Exception as err:
            raise ExportToS3Error(err)

    def get_export_progress(self, task_id: str) -> str:
        try:
            response = self.client.describe_export_tasks(taskId=task_id)
            status = response['exportTasks'][0]['status']['code']
            return status
        except Exception as err:
            raise GetExportTaskError(err)

    @classmethod
    def finishes(cls, status_code: str) -> bool:
        """Stellen Sie anhand des Statuscodes fest, ob die Exportaufgabe abgeschlossen wurde

        Args:
            status_code (str):
                describe_export_Statuscode in der Antwort der Aufgabe enthalten

        Raises:
            ExportToS3Failure:Wenn der Statuscode ABGESAGT oder FEHLGESCHLAGEN ist

        Returns:
            bool
        """

        uppercase_status_code = status_code.upper()

        if uppercase_status_code == 'COMPLETED':
            return True
        elif uppercase_status_code in ['CANCELLED', 'FAILED']:
            raise ExportToS3Failure('Exportfehler nach S3')
        return False

main

Verwenden Sie LogGroup. \ _ \ _ Subclasses \ _ \ _ (), um die festgelegte untergeordnete Klasse der zu exportierenden Protokollgruppe abzurufen. \ _ \ _ Unterklassen \ _ \ _ () gibt eine Liste zurück, also drehen Sie sie mit einer for-Anweisung. Sie können jeweils nur eine CloudWatch Logs-Exportaufgabe in Ihrem Konto ausführen. Klicken Sie daher auf die API "description_export_tasks", um festzustellen, ob die Aufgabe abgeschlossen ist. Wenn es nicht abgeschlossen ist, werde ich auf 5s warten. Da create_export_task eine asynchrone API ist, haben wir keine andere Wahl, als sie auf diese Weise abzufragen.

def export_to_s3(exporter: Exporter, target: Type[LogGroup]) -> bool:
    task_id = exporter.export(target)
    logger.info(f'{target.log_group}Wird nach S3 exportiert({task_id=})')

    while True:
        status = exporter.get_export_progress(task_id)
        if exporter.finishes(status):
            return True
        sleep(5)


def main(event: Any, context: Any) -> bool:
    exporter = Exporter(region='ap-northeast-1')
    targets = LogGroup.__subclasses__()
    logger.info(f'Die zu exportierende Protokollgruppe ist{len(targets)}Stücke')

    for target in targets:
        try:
            export_to_s3(exporter, target)
        except GetExportTaskError as err:
            logger.warning(err)
            logger.warning(f'{target.log_group}Fehler beim Fortschreiten')
        except Exception as err:
            logger.error(err)
            logger.error(f'{target.log_group}Export nach S3 fehlgeschlagen')
        else:
            logger.info(f'{target.log_group}Export nach S3')

    return True

serverless.yml

Das Folgende ist ein Teilauszug. Stellen Sie die IAM-Rolle so ein, dass Lambda Exportaufgaben erstellen und Aufgabeninformationen abrufen kann. Dann möchte ich den Export jeden Tag um JST 14:00 ausführen, also geben Sie für Ereignisse cron (0 5 * *? *) An. CloudWatch-Ereignisse werden unter UTC ausgeführt. Wenn Sie also -9h ausführen, wird es wie erwartet um 14:00 Uhr JST ausgeführt.

  iamRoleStatements:
    - Effect: 'Allow'
      Action:
        - 'logs:createExportTask'
        - 'logs:DescribeExportTasks'
      Resource:
        - 'arn:aws:logs:${self:provider.region}:${self:custom.accountId}:log-group:*'

functions:
  export:
    handler: src/handler.main
    memorySize: 512
    timeout: 120
    events:
      - schedule: cron(0 5 * * ? *)
    environment:
      TZ: Asia/Tokyo

abschließend

homoluctus / lambda-cwlogs-s3 verfügt auch über eine CloudFormation-Vorlage zum Erstellen von GitHub-Aktionen und Ziel-S3s. Bitte nehmen Sie Bezug darauf.

Reference

Recommended Posts

[Python] Exportieren Sie regelmäßig mit Lambda aus CloudWatch-Protokollen nach S3
Stellen Sie mit AWS Lambda Python eine Verbindung zu s3 her
Verschieben Sie CloudWatch-Protokolle regelmäßig mit Lambda nach S3
[Lambda] [Python] Von Lambda auf Twitter posten!
Ich möchte Protokolle mit Python analysieren
Zugriff auf RDS von Lambda (Python)
Laden Sie das, was Sie angefordert haben, mit AWS Lambda Python in S3 hoch
Kopieren Sie Daten von Amazon S3 mit Python (boto) in Google Cloud Storage.
Führen Sie das WEB-Scraping regelmäßig mit AWS-Lambda + Python + Cron aus
Ich habe versucht, CloudWatch-Daten mit Python abzurufen
Beispiel für eine Slack-Benachrichtigung mit Python Lambda
Laden Sie Dateien mit Lambda (Python) auf Google Drive hoch.
Von der Python-Umgebungskonstruktion zur virtuellen Umgebungskonstruktion mit Anaconda
Änderungen von Python 3.0 zu Python 3.5
So kratzen Sie Bilddaten von Flickr mit Python
Vom Kauf eines Computers bis zur Ausführung eines Programms auf Python
Ich möchte Lambda mit Python auf Mac AWS!
Kopieren Sie S3-Dateien mit GSUtil von Python nach GCS
ODBC-Zugriff auf SQL Server von Linux mit Python
Lambda-Funktion (Python-Version), die Elemente dekomprimiert und in CloudWatch-Protokolle ausgibt, wenn eine komprimierte Datei auf s3 hochgeladen wird
Stellen Sie mit Python eine Verbindung zu BigQuery her
Post von Python nach Slack
[S3] CRUD mit S3 unter Verwendung von Python [Python]
Flirte von PHP nach Python
Stellen Sie mit Python eine Verbindung zu Wikipedia her
[AWS] Versuchen Sie, die Python-Bibliothek mit SAM + Lambda (Python) zur Ebene hinzuzufügen.
Post to Slack mit Python 3
S3-Betrieb mit Python Boto3
Anaconda aktualisiert von 4.2.0 auf 4.3.0 (python3.5 aktualisiert auf python3.6)
Fragen Sie Athena von Lambda Python ab
Herausforderung Problem 5 mit Python: Lambda ... Ich habe mich entschieden, ohne zu kopieren
Einführung in Python für VBA-Benutzer - Aufrufen von Python aus Excel mit xlwings-
Wechseln Sie von Python2.7 zu Python3.6 (centos7)
Stellen Sie von Python aus eine Verbindung zu SQLite her
Schalten Sie Python mit Alternativen auf 2.7 um
Schreiben Sie mit Python in csv
Mit Skype benachrichtigen Sie mit Skype von Python!
Wie benutzt man Python Lambda?
Verarbeiten Sie die mit Redshift entladene gzip-Datei mit Python of Lambda, gzipen Sie sie erneut und laden Sie sie in S3 hoch
Eine Geschichte, die ich behoben habe, als ich das Lambda-Protokoll von Cloudwatch Logs erhalten habe
[Python 3.8 ~] Wie man rekursive Funktionen mit Lambda-Ausdrücken intelligent definiert
Senden Sie mit ESP32-WROOM-32 aufgenommene Bilder an AWS (API Gateway → Lambda → S3).
Versuchen Sie, mit Python3 eine Zeichenfolge aus einem Bild zu extrahieren
Vorsichtsmaßnahmen beim Ausführen von Python unter EC2 über AWS Lambda (Befehl ausführen)
Einführung in die Datenanalyse mit Python P17-P26 [ch02 1.usa.gov Daten von bit.ly]
Bearbeiten von Kintondaten mit dem Python & C Data ODBC-Treiber von AWS Lambda
Ich las "Das Lernen mit Python von der Einführung bis zur Praxis stärken", Kapitel 1
Von der Einführung von JUMAN ++ bis zur morphologischen Analyse von Japanisch mit Python
Übergeben Sie die Liste von Python an C ++ als Referenz in pybind11
Ich las "Das Lernen mit Python von der Einführung bis zur Praxis stärken", Kapitel 2
HDA-Verteilung von Houdini zum Exportieren von FBX mit Hierarchie und Transformationen
Versuchen Sie, CloudWatch-Metriken mit der Python-Datenquelle re: dash abzurufen
Der schnellste Weg, um regelmäßig Kamerabilder mit Pythons OpenCV zu erhalten
[Python] Versuchen Sie, Zeichen aus Bildern mit OpenCV und pyocr zu erkennen
Rufen Sie Matlab von Python zur Optimierung auf
[Einführung in die Udemy Python3 + -Anwendung] 58. Lambda
Python: So verwenden Sie Async mit
Rufen Sie C von Python mit DragonFFI auf
Verwenden von Rstan aus Python mit PypeR