通过 Cloudwatch 实现用户日志记录自动化

Question

我的公司有这个任务，我必须通过 cloudwatch 每月进行一次用户访问审查。

这是一个手动过程，我必须转到 cloudwatch > cloudwatch_logs > log_groups > /var/log/example_access > example-instance，然后记录随机生成日期的用户列表的日志。示例实例是一个证书管理器框，它链接到我们的整个生产车队节点。我还必须记录用户在特定节点上使用的命令。

想知道有什么方法可以使这个过程自动化并将其转储到 word 文档中？随着用户/员工列表的增加，它变得越来越痛苦。谢谢

Answer 1

当然有，不过我不认为您需要 Word 文档，我会在 AWS 上启动一个 elasticsearch 实例，然后为需要数据 Kibana 的用户提供访问权限。

在一个组织中循环的 word 文档也很糟糕，这取决于你的 windows/office 版本，它会带来风险。

添加此 lambda 函数，然后进入 cloudwatch 并将其作为订阅过滤器添加到正确的日志组。

请注意，如果日志条目不是以 JSON 格式记录或具有时髦的格式，则可能会丢失日志条目，如果您使用的是标准日志格式，它应该可以工作。

/* eslint-disable */
// Eslint disabled as this is adapted AWS code.

const zlib = require('zlib')
const elasticsearch = require('elasticsearch')

/**
 * This is an example function to stream CloudWatch logs to ElasticSearch.
 * @param event
 * @param context
 * @param callback
 * @param utils
 */
export default (event, context, callback) => {
    context.callbackWaitsForEmptyEventLoop = true

    const payload = new Buffer(event.awslogs.data, 'base64')

    const esClient = new elasticsearch.Client({
        httpAuth: process.env.esAuth, // your params here
        host: process.env.esEndpoint, // your params here.
    })

    zlib.gunzip(payload, (err, result) => {

        if (err) {
            return callback(null, err)
        }

        const logObject = JSON.parse(result.toString('utf8'))

        const elasticsearchBulkData = transform(logObject)

        const params = { body: [] }
        params.body.push(elasticsearchBulkData)

        esClient.bulk(params, (err, resp) => {

            if (err) {
        callback(null, 'success')
        return
    }

        })

        callback(null, 'success')
    })
}

function transform(payload) {
    if (payload.messageType === 'CONTROL_MESSAGE') {
        return null
    }

    let bulkRequestBody = ''

    payload.logEvents.forEach((logEvent) => {
        const timestamp = new Date(1 * logEvent.timestamp)

        // index name format: cwl-YYYY.MM.DD
        const indexName = [
            `cwl-${process.env.NODE_ENV}-${timestamp.getUTCFullYear()}`,              // year
            (`0${timestamp.getUTCMonth() + 1}`).slice(-2),  // month
            (`0${timestamp.getUTCDate()}`).slice(-2),          // day
        ].join('.')

        const source = buildSource(logEvent.message, logEvent.extractedFields)
        source['@id'] = logEvent.id
        source['@timestamp'] = new Date(1 * logEvent.timestamp).toISOString()
        source['@message'] = logEvent.message
        source['@owner'] = payload.owner
        source['@log_group'] = payload.logGroup
        source['@log_stream'] = payload.logStream

        const action = { index: {} }
        action.index._index = indexName
        action.index._type = 'lambdaLogs'
        action.index._id = logEvent.id

        bulkRequestBody += `${[
            JSON.stringify(action),
            JSON.stringify(source),
        ].join('\n')}\n`
    })
    return bulkRequestBody
}

function buildSource(message, extractedFields) {
    if (extractedFields) {
        const source = {}

        for (const key in extractedFields) {
            if (extractedFields.hasOwnProperty(key) && extractedFields[key]) {
                const value = extractedFields[key]

                if (isNumeric(value)) {
                    source[key] = 1 * value
                    continue
                }

                const jsonSubString = extractJson(value)
                if (jsonSubString !== null) {
                    source[`$${key}`] = JSON.parse(jsonSubString)
                }

                source[key] = value
            }
        }
        return source
    }

    const jsonSubString = extractJson(message)
    if (jsonSubString !== null) {
        return JSON.parse(jsonSubString)
    }

    return {}
}

function extractJson(message) {
    const jsonStart = message.indexOf('{')
    if (jsonStart < 0) return null
    const jsonSubString = message.substring(jsonStart)
    return isValidJson(jsonSubString) ? jsonSubString : null
}

function isValidJson(message) {
    try {
        JSON.parse(message)
    } catch (e) { return false }
    return true
}

function isNumeric(n) {
    return !isNaN(parseFloat(n)) && isFinite(n)
}

现在您应该将日志放入 elastic 中，进入 Kibana，您可以按日期搜索，甚至可以编写端点以允许人们查询自己的数据！

简单的方法是让利益相关者访问 Kibana 并让他们查看。

可能不是你想要的，我认为它会更好。