Node.jsで常時フォルダを監視してS3へファイルをアップロードする

概要

あるフォルダに不定期にCSVファイルが作成されるので、そのCSVファイルをS3にアップロードするプログラムをNode.jsで作成します。
処理に失敗するとAWS SESを使って、管理者へメールを配信します。
loggerも使って何が起こったかを追えるようにします。
全コードはこちら

フロー

起動時に監視フォルダとローカルの保存フォルダが存在するかチェック

initial_check.js

const fs = require("fs");
const fsPromises = fs.promises;

const initialCheck = async (WATCHING_DIR, DEST_DIR) => {
  try {
    await fsPromises.access(
      WATCHING_DIR,
      fs.constants.R_OK | fs.constants.W_OK
    );
    await fsPromises.access(DEST_DIR, fs.constants.R_OK | fs.constants.W_OK);
  } catch (error) {
    throw new Error("監視/管理フォルダへアクセスできない");
  }
};

module.exports = initialCheck;

指定のフォルダが存在しなければ、プログラムは終了。その際にAWS SESから管理者へメールが配信されるようにしてあります。

chokidarでwatcherを起動して正常動作をチェック

watcherの設定

watcher.js

const chokidar = require("chokidar"); // フォルダ監視用
const WATCHING_DIR = require("./config.json").WATCHING_DIR;

// Initialize watcher.
const watcher = chokidar.watch(WATCHING_DIR, {
  ignored: /[\/\\]\./,
  persistent: true,
  usePolling: true,
  interval: 10000
});

module.exports = watcher;

watcherの動作確認

monitoring.js

  const watcher = require("./watcher"); // フォルダ監視用

  watcher.on("ready", async () => {
    await logger.info("Initial scan complete. Ready for changes");
    const watchedPaths = watcher.getWatched();
    await logger.info("watchedPaths :", watchedPaths);
  });

監視フォルダにファイルが追加されたときの動作を設定

monitoring.js

  // ファイルの追加を検知
  watcher.on("add", async filePath => {
    await logger.info("add file: ", filePath);
    try {
      await bucketExistCheck();
      await logger.info("Bucket Existed.");
      await fileCopyUploadDelete(filePath);
    } catch (error) {
      await errorState(error);
    }
  });

S3のバケットが存在することを確認してから追加されたファイルをS3へアップロードします。

watcherがエラーとなった場合

monitoring.js

  watcher.on("error", async error => {
    await errorState(error);
    await watcher
      .close()
      .then(() => logger.info("Watcher closed: watcher on error"));
  });

S3バケットが存在するかをチェック

monitoring.js

const bucketExistCheck = async () => {
  try {
    await s3.headBucket({ Bucket: BUCKET }).promise();
  } catch (error) {
    throw new Error("S3 Bucket not exist!");
  }
};

追加されたファイルを処理する流れ

1. ローカルで保存するためフォルダにYYYYMMの名前でフォルダを作成
2. そのフォルダに追加されたファイルをコピー
3. 追加されたファイルをS3へアップロードするためのパラメーターを作成
4. S3へアップロード
5. 監視フォルダから追加されたファイルを削除

monitoring.js

const fileCopyUploadDelete = async filePath => {
  const filenameParse = path.parse(filePath);
  // copyする前にYYYYMMのフォルダが存在するか確認して、存在しなければフォルダ作成する
  const sendPathThisMonthDir = await mkdirThisMonth();
  const destFilePath = path.join(sendPathThisMonthDir, filenameParse.base);
  await fsPromises.copyFile(filePath, destFilePath, COPYFILE_EXCL);
  await logger.info("File copy success!", filePath, destFilePath);
  // コピー失敗した場合は？ => 'EEXIST: file already exists, copyfile'のエラーメッセージthrow
  // COPYFILE_EXCLを指定しているためコピー先に同名ファイルがあった場合は上記エラー
  // renameメソッドでもファイル移動が可能だが、移動先に同名ファイルが存在する場合に上書きとなってしまうため利用せず

  const uploadParams = await createUploadParams(filePath);
  await logger.info("File Read Success", filePath);

  const data = await s3.putObject(uploadParams).promise();
  await logger.info("Upload Success", data);

  await fsPromises.unlink(filePath);
  await logger.info("File delete Success!", filePath);
};

YYYYMMのフォルダが存在しなければ作成する↓

monitoring.js

const mkdirThisMonth = async () => {
  try {
    const today = new Date();
    const monthMM = ("0" + (today.getMonth() + 1)).slice(-2);
    const yyyymm = today.getFullYear().toString() + monthMM;
    const dirPath = path.join(DEST_DIR, yyyymm);
    await fsPromises.mkdir(dirPath);
    logger.info("Make Directory", dirPath);
    return dirPath;
  } catch (error) {
    if (error.code === "EEXIST") {
      logger.warn(error.message);
      return error.path;
    }
    throw new Error(error);
  }
};

S3へアップロードするためのパラメーターを作成↓
ファイルはcsvからjsonへ変換。なんとなくファイル名にランダム文字列を追加。
ContentMD5キーを指定して正しくアップロードされたかをチェック

monitoring.js

const createUploadParams = async filePath => {
  try {
    const body = await orderCsvToJson(filePath); // ファイルの中身をjson形式へ変換
    const md5hash = crypto.createHash("md5");
    const md5sum = md5hash.update(body).digest("base64");
    const randomString = crypto.randomBytes(8).toString("hex");
    // ファイル名が重複しないようにする
    const filenameParse = path.parse(filePath);
    const key = filenameParse.name + "_" + randomString + ".json";
    return {
      Bucket: BUCKET,
      Key: key,
      Body: body,
      ContentMD5: md5sum
    };
  } catch (error) {
    throw new Error(error);
  }
};

エラーが発生した場合はSESでメール配信

monitoring.js

const errorState = async error => {
  await logger.error(error.message);
  console.error(error);
  // 管理者へメール配信
  const sesParams = createSESParams(error);
  await sesSendMail(sesParams)
    .then(res => {
      logger.info("管理者へメール配信", res);
    })
    .catch(error => {
      logger.error("メール配信エラー", error);
    });
};

最後に

おおまかに書きました。
詳細はGitHubをご確認ください。
もっと簡単な方法があるのだと思います。