0
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

@nabeo_ji(なべお)

[AWS]EOL情報の自動収集（endoflife.date + EventBridge Scheduler + Lambda + S3）

0
Posted at

当方、AWS、OCIのインフラエンジニア（３年）をしつつ、過去はアプリのカスタマイズ開発を６年、ホスト系では５年と経験してきましたが、
仕事の中でラクできるところはどんどん自動化をする、自動化狂いでした。

日頃の仕事を楽にするためにも、当該記事のようなチョイっと作れるツールはどんどん導入していきたいですね。
ということで本題です。

目的

インフラ運用をしていると、OS / ミドルウェア / DB の EOL（End of Life） を把握しておくことは避けて通れません。

  • EOLを過ぎたバージョンを使い続けると、セキュリティパッチが提供されない
  • アップグレード計画を立てるためにも「いつ切れるか」を一覧で持っておきたい
  • 監査・脆弱性対応・パッチ運用の場面でもEOLの証跡が求められがち

そこでこの記事では、endoflife.date （参考：https://endoflife.date/ ）の情報を使って EOL情報を定期的に取得し、S3にCSV/JSONで蓄積する仕組みを作ります。

全体構成

  • EventBridge Scheduler：定期実行（例：毎日朝に収集）
  • Lambda：endoflife.date API v1からEOL情報を取得し、CSV/JSONに整形してS3へ保存
  • S3：inventories/eol/ 配下に eol_inventory_YYYYmmddTHHMMSSZ.csv/json として蓄積

事前準備

1. S3バケット

例：s3-eol-list

2. Lambda実行ロール（最小権限）

LambdaからS3にPutObjectするために、実行ロールに以下のようなポリシーを付けます。

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Sid": "PutOnlyToEolPrefix",
      "Effect": "Allow",
      "Action": ["s3:PutObject"],
      "Resource": "arn:aws:s3:::s3-eol-list/inventories/eol/*"
    }
  ]
}

3. endoflife.date の製品名（slug）一覧化

endoflife.date のAPI仕様（ https://endoflife.date/docs/api/v1/# ）をもとに、製品名を下調べします。
例えば、Amazon LinuxのEOL情報を確認できるWebページは下記です。
https://endoflife.date/amazon-linux
APIは下記です。
https://endoflife.date/api/v1/products/amazon-linux
APIが機能するか確認するには、Linuxだと下記のようなコマンドになります。

curl -X 'GET' 'https://endoflife.date/api/v1/products/amazon-linux' -H 'accept: application/json'

APIは製品ごとにAliasがあったりします。URIはAPI仕様を確認できるページで検索できるので、事前に製品ごとのURIを確認しておきましょう。
この製品名のことをソースコード内で”slug”に指定してAPIをたたいています。

ソースコード（Lambda用）

Pythonランタイムに標準で入っている urllib を使っているので、追加ライブラリ不要です。
（boto3はLambdaにプリインストールされています）

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

"""
AWS Lambda: endoflife.date から EOL 情報を取得し、CSV/JSON を S3 に保存する。

- 定期実行（EventBridge Scheduler）を想定：入力 event は空 {} でOK
- 固定値（S3保存先など）は環境変数で管理する想定
- endoflife.date API v1（/api/v1/products/{slug}/）のレスポンスを主対象として処理する

必須の環境変数:
  - BUCKET: 出力先S3バケット
  - PREFIX: 出力先キーのプレフィックス（例: inventories/eol）

任意の環境変数:
  - PRODUCTS: 取得対象 slug のカンマ区切り（未指定なら DEFAULT_PRODUCTS を使用）
      例: apache-http-server,mssqlserver,tomcat,amazon-linux
  - WRITE_LATEST: true/false（trueなら latest.csv/json も上書き作成）
  - TIMEOUT_SEC: 外部HTTPタイムアウト秒（例: 5）
  - BASE_URL: APIベースURL（デフォルト: https://endoflife.date/api/v1）
  - SSE: S3のServerSideEncryption（例: AES256 または aws:kms）

設定値の優先順位（上書きルール）
- 基本方針：イベント（event）で指定した値が最優先。次に環境変数（env）。
- event / env のいずれにも無い場合は、コード内のデフォルト値を使用する。
"""

from __future__ import annotations

import csv
import datetime as dt
import io
import json
import os
import urllib.error
import urllib.request
from typing import Any, Dict, List, Optional, Tuple

import boto3

DEFAULT_PRODUCTS: List[Tuple[str, str]] = [
    ("Apache HTTP Server", "apache-http-server"),
    ("Microsoft SQL Server", "mssqlserver"),
    ("Apache Tomcat", "tomcat"),
    ("Amazon Linux", "amazon-linux"),
]

FIELDNAMES = [
    "product",
    "slug",
    "cycle",
    "releaseDate",
    "eol",
    "support",
    "latest",
    "latestReleaseDate",
    "lts",
    "discontinued",
    "link",
    "raw",
]


def utc_now_compact() -> str:
    return dt.datetime.now(dt.timezone.utc).strftime("%Y%m%dT%H%M%SZ")


def safe_str(v: Any) -> str:
    if v is None:
        return ""
    if isinstance(v, bool):
        return "true" if v else "false"
    return str(v)


def env_bool(name: str, default: bool = False) -> bool:
    raw = (os.getenv(name) or "").strip().lower()
    if raw == "":
        return default
    return raw in ("1", "true", "yes", "y", "on")


def load_config(event: Any) -> Dict[str, Any]:
    event = event if isinstance(event, dict) else {}

    bucket = event.get("bucket") or os.getenv("BUCKET")
    prefix = event.get("prefix") or os.getenv("PREFIX")
    if not bucket or not prefix:
        raise ValueError("Missing required config. Set env BUCKET and PREFIX (event is optional).")

    region = event.get("region") or os.getenv("AWS_REGION") or os.getenv("AWS_DEFAULT_REGION") or "ap-northeast-1"
    base_url = event.get("base_url") or os.getenv("BASE_URL") or "https://endoflife.date/api/v1"
    timeout_sec = int(event.get("timeout_sec") or os.getenv("TIMEOUT_SEC") or 10)
    write_latest = bool(event.get("write_latest") or env_bool("WRITE_LATEST", default=False))
    sse = (event.get("sse") or os.getenv("SSE") or "").strip() or None

    products_raw = event.get("products")
    if products_raw is None:
        products_raw = os.getenv("PRODUCTS")

    if isinstance(products_raw, list):
        slugs = [str(x).strip() for x in products_raw if str(x).strip()]
    elif isinstance(products_raw, str) and products_raw.strip():
        slugs = [s.strip() for s in products_raw.split(",") if s.strip()]
    else:
        slugs = [slug for _, slug in DEFAULT_PRODUCTS]

    return {
        "bucket": bucket,
        "prefix": prefix,
        "region": region,
        "base_url": base_url,
        "timeout_sec": timeout_sec,
        "write_latest": write_latest,
        "sse": sse,
        "slugs": slugs,
    }


def http_get_json(url: str, timeout_sec: int) -> Any:
    req = urllib.request.Request(
        url,
        headers={"Accept": "application/json", "User-Agent": "eol-to-s3-lambda/1.0"},
        method="GET",
    )
    try:
        with urllib.request.urlopen(req, timeout=timeout_sec) as resp:
            return json.loads(resp.read().decode("utf-8"))
    except urllib.error.HTTPError as e:
        raise RuntimeError(f"HTTP error {e.code} for {url}: {e.reason}") from e
    except urllib.error.URLError as e:
        raise RuntimeError(f"URL error for {url}: {e.reason}") from e
    except json.JSONDecodeError as e:
        raise RuntimeError(f"JSON decode error for {url}: {e}") from e


def v1_to_cycle_list(v1_data: Dict[str, Any], slug: str) -> List[Dict[str, Any]]:
    result = v1_data.get("result")
    if not isinstance(result, dict):
        raise ValueError(f"Unexpected v1 response for {slug}: missing result")

    releases = result.get("releases")
    if not isinstance(releases, list):
        raise ValueError(f"Unexpected v1 response for {slug}: missing releases[]")

    product_html = (result.get("links") or {}).get("html")
    last_modified = v1_data.get("last_modified")

    cycles: List[Dict[str, Any]] = []
    for rel in releases:
        if not isinstance(rel, dict):
            continue

        latest = rel.get("latest") or {}
        is_eol = bool(rel.get("isEol"))
        eol_from = rel.get("eolFrom")

        cycles.append(
            {
                "cycle": rel.get("name") or rel.get("label"),
                "releaseDate": rel.get("releaseDate"),
                "eol": eol_from if is_eol else False,
                "support": rel.get("isMaintained"),
                "latest": latest.get("name"),
                "latestReleaseDate": latest.get("date"),
                "lts": rel.get("isLts"),
                "discontinued": False,
                "link": latest.get("link") or product_html,
                "_v1_last_modified": last_modified,
                "_v1_release": rel,
            }
        )

    return cycles


def fetch_product_cycles(base_url: str, slug: str, timeout_sec: int) -> List[Dict[str, Any]]:
    url = f"{base_url.rstrip('/')}/products/{slug}/"
    data = http_get_json(url, timeout_sec)

    if isinstance(data, dict):
        return v1_to_cycle_list(data, slug)

    if isinstance(data, list):
        return data

    preview = str(data)[:200]
    raise ValueError(f"Unexpected API response shape for {slug}: got {type(data)} preview={preview}")


def normalize_rows(products: List[Tuple[str, str]], cycles_by_product: Dict[str, List[Dict[str, Any]]]) -> List[Dict[str, Any]]:
    rows: List[Dict[str, Any]] = []
    for display_name, slug in products:
        for c in cycles_by_product.get(slug, []):
            if not isinstance(c, dict):
                continue
            rows.append(
                {
                    "product": display_name,
                    "slug": slug,
                    "cycle": safe_str(c.get("cycle")),
                    "releaseDate": safe_str(c.get("releaseDate")),
                    "eol": safe_str(c.get("eol")),
                    "support": safe_str(c.get("support")),
                    "latest": safe_str(c.get("latest")),
                    "latestReleaseDate": safe_str(c.get("latestReleaseDate")),
                    "lts": safe_str(c.get("lts")),
                    "discontinued": safe_str(c.get("discontinued")),
                    "link": safe_str(c.get("link")),
                    "raw": json.dumps(c, ensure_ascii=False, sort_keys=True),
                }
            )
    rows.sort(key=lambda x: (x["product"], x["cycle"], x["releaseDate"]))
    return rows


def rows_to_csv_bytes(rows: List[Dict[str, Any]]) -> bytes:
    buf = io.StringIO()
    w = csv.DictWriter(buf, fieldnames=FIELDNAMES, extrasaction="ignore", lineterminator="\n")
    w.writeheader()
    for r in rows:
        w.writerow({k: safe_str(r.get(k)) for k in FIELDNAMES})
    return buf.getvalue().encode("utf-8")


def rows_to_json_bytes(rows: List[Dict[str, Any]]) -> bytes:
    return json.dumps(rows, ensure_ascii=False, indent=2).encode("utf-8")


def s3_put_object(s3_client, bucket: str, key: str, body: bytes, content_type: str, sse: Optional[str]) -> None:
    kwargs: Dict[str, Any] = {"Bucket": bucket, "Key": key, "Body": body, "ContentType": content_type}
    if sse:
        kwargs["ServerSideEncryption"] = sse
    s3_client.put_object(**kwargs)


def run_job(cfg: Dict[str, Any]) -> Dict[str, Any]:
    slug_to_display = {slug: name for name, slug in DEFAULT_PRODUCTS}
    products: List[Tuple[str, str]] = [(slug_to_display.get(slug, slug), slug) for slug in cfg["slugs"]]

    cycles_by_product: Dict[str, List[Dict[str, Any]]] = {}
    for _, slug in products:
        cycles_by_product[slug] = fetch_product_cycles(cfg["base_url"], slug, cfg["timeout_sec"])

    rows = normalize_rows(products, cycles_by_product)

    csv_bytes = rows_to_csv_bytes(rows)
    json_bytes = rows_to_json_bytes(rows)

    s3 = boto3.client("s3", region_name=cfg["region"])
    ts = utc_now_compact()
    prefix = cfg["prefix"].strip("/")

    csv_key = f"{prefix}/eol_inventory_{ts}.csv"
    json_key = f"{prefix}/eol_inventory_{ts}.json"

    s3_put_object(s3, cfg["bucket"], csv_key, csv_bytes, "text/csv; charset=utf-8", cfg["sse"])
    s3_put_object(s3, cfg["bucket"], json_key, json_bytes, "application/json; charset=utf-8", cfg["sse"])

    if cfg["write_latest"]:
        s3_put_object(s3, cfg["bucket"], f"{prefix}/latest.csv", csv_bytes, "text/csv; charset=utf-8", cfg["sse"])
        s3_put_object(s3, cfg["bucket"], f"{prefix}/latest.json", json_bytes, "application/json; charset=utf-8", cfg["sse"])

    return {
        "status": "ok",
        "bucket": cfg["bucket"],
        "keys": [csv_key, json_key] + ([f"{prefix}/latest.csv", f"{prefix}/latest.json"] if cfg["write_latest"] else []),
        "rows": len(rows),
    }


def lambda_handler(event, context):
    cfg = load_config(event)
    print(json.dumps({"msg": "start", "bucket": cfg["bucket"], "prefix": cfg["prefix"], "slugs": cfg["slugs"]}, ensure_ascii=False))
    result = run_job(cfg)
    print(json.dumps({"msg": "done", **result}, ensure_ascii=False))
    return result

Lambdaに設定する環境変数

最低限必要：

  • BUCKET：出力先バケット名
  • PREFIX：出力先プレフィックス
    ※イベントのほうにも仕込めるようにしてますが、まず簡単に動かすなら上記の環境変数を設定すればデフォルト値で動くようにしてます。

例：

  • BUCKET=s3-eol-list
  • PREFIX=inventories/eol

任意で指定できるもの：

  • PRODUCTS：取得対象slug（カンマ区切り）
  • WRITE_LATEST：trueなら latest.csv/json を上書き生成
  • TIMEOUT_SEC：HTTPタイムアウト秒
  • BASE_URL：APIベースURL（通常は不要）
  • SSE：S3のServerSideEncryption（例：AES256）

設定値の優先順位（イベント / 環境変数 / デフォルト）

このLambdaは 定期実行ではイベントを空 {} にして運用できるように作っています。
一方で、手動実行やテストでは event を渡して挙動を変えることもできます。

基本ルール

  • 基本方針：イベント（event）で指定した値が最優先。次に環境変数（env）
  • イベント / 環境変数 のいずれにも無い場合は、コード内のデフォルト値を使用する
設定項目 説明（日本語） デフォルト
bucket（必須） 出力先S3バケット名 ※未設定はエラー（指定サンプル：s3-eol-list）
prefix（必須） 出力先S3キーのプレフィックス（フォルダ相当） ※未設定ならエラー（指定サンプル：inventories/eol）
region AWSリージョン（boto3のS3クライアント作成に使用） "ap-northeast-1"
base_url endoflife.date APIのベースURL "https://endoflife.date/api/v1"
timeout_sec 外部HTTPリクエストのタイムアウト秒（Lambda全体のTimeoutとは別） 10
write_latest latest.csv/json を同一キーで上書き出力するか False
（※現行実装は event.write_latest=False の場合でも、env.WRITE_LATEST=True だと True になり得る）
sse S3 PutObject時のServer-Side Encryption指定（未指定ならSSEヘッダ無し） None
（未指定ならSSEヘッダ無しでPutObject。AES256 など指定）
products 取得対象プロダクトのslug一覧（表示名はDEFAULT_PRODUCTSから補完、無ければslugのまま） コード内記述のDEFAULT_PRODUCTSに指定されたデフォルト一覧
・event.products が list: ["apache-http-server", "tomcat"]（slug配列）
・event.products が str: "apache-http-server,tomcat"（カンマ区切り）

event の全項目指定サンプル（テストイベント例）

以下は、Lambdaテストで「event側ですべての設定値を上書き」したいときの例。

{
  "bucket": "s3-eol-list",
  "prefix": "inventories/eol",
  "region": "ap-northeast-1",
  "base_url": "https://endoflife.date/api/v1",
  "timeout_sec": 5,
  "write_latest": true,
  "sse": "AES256",
  "products": [
    "apache-http-server",
    "mssqlserver",
    "tomcat",
    "amazon-linux"
  ]
}

EventBridge Schedulerで定期起動する

EventBridge Scheduler の「ターゲット」にLambdaを指定し、入力（Input）は {} でOKです。

  • 取得対象や保存先を変えたい場合は、環境変数を更新する運用がシンプルです
  • WRITE_LATEST=true にすれば、参照する側は常に latest.csv/json を見るだけで済みます

ぶっちゃけ

超絶シンプルに実装するなら、下記。

echo "" > /tmp/eollist.txt
curl -X 'GET' "https://endoflife.date/api/v1/products/amazon-linux"   -H 'accept: application/json' >> /tmp/eollist.txt
curl -X 'GET' "https://endoflife.date/api/v1/products/tomcat"         -H 'accept: application/json' >> /tmp/eollist.txt
curl -X 'GET' "https://endoflife.date/api/v1/products/zabbix"         -H 'accept: application/json' >> /tmp/eollist.txt
curl -X 'GET' "https://endoflife.date/api/v1/products/mssqlserver"    -H 'accept: application/json' >> /tmp/eollist.txt

製品名をテキスト格納しておいて、配列に取得してループかけてもいいですね。
300行以上のLambdaは何だったのか？　って感じですが趣味ですね。はい。
いままでプログラミングを十何年もやってましたが、その間、作ってきたツールって、だいたいbashだったり、Linux OSのワンライナーで事足りてしまうことが多くて、あんまり派手なツール作ることは少なくなってきたように思います。

まとめ

  • なんでもいいからAPIのURIたたいて情報取得して何かに集積すればいい
0
0
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
Sign upLogin
0
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?