How to Create Lambda function for pushing logs to es

Hi Folks !So, I hope you all are well good to see you again. I will show you a new post which is very interesting and also helps you to learn new thing about ELK Stack. So, In this blog we’ll see that How to Create Lambda function for pushing logs to elastic-search.

In this github user pushes the code and github actions push the data in the csv format to the s3 bucket ,which will further trigger the lambda and after that will connects with the Elasticsearch and pushes the logs to it and can visualize them Kibana. So, this is how we can cretae Function for pushing logs ES.

let’s get started !!!

Overview

We need an efficient way to push logs to the Elasticsearch and visualize them in Kibana because of efficient way of working . So, we’ll be using AWS Lambda Function to send S3 Bucket uploaded logs in CSV File to ElasticSearch and Visualize them.

So, the following workflow let’s you know the exact working of the above function:

Once the User push the data to the github repository.

Github Repository uses github actions to push the github logs in CSV file format to the AWS S3 Bucket.

S3 Bucket trigger the AWS Lambda that will connect with the Elasticsearch.

Visualize the github commit and pr details in Kibana.

User: Github User will push the code or logs to the github repository.

Repository: Repositories in GIT contain a collection of files of various different versions of a Project.

Github Actions: GitHub Actions is a continuous integration and continuous delivery (CI/CD) platform that allows you to automate your build, test, and deployment pipeline.

AWS S3 Bucket: An Amazon S3 bucket is a public cloud storage resource available in AWS. Indeed S3 is an object storage offering. So, Amazon S3 buckets, are similar to file folders, store objects, consists of data and its descriptive metadata.

Lambda Function: Lambda is a serverless computing service provided by Amazon Web Services (AWS) that lets you run your code without provisioning or managing servers.

Elasticsearch: It is the central component of the Elastic Stack, because of free and open tools for data ingestion, enrichment, storage, analysis, and visualization.

Kibana: This is an free and open frontend application that sits on top of the Elastic Stack, because it provides search and data visualization capabilities for data indexed in Elasticsearch.

Prerequisites

you should neet at least these two prerequisite for the flow.

AWS S3 Bucket

Iam role (AWS S3 full access, Lambda full access, Cloudwatch full access )

Steps for pushing S3 logs to the Elasticsearch through Lambda

So, let’s see the steps of How to Create Lambda function for pushing logs to elastic-search one by one.

Create an AWS Lambda function.

• Open your AWS Console and search for service as Lambda
• Go to Lambda >> Function >> create Function >> select Author From Scratch >> Mention function name >> select Python3.7 Runtime >> Create Function

 Add the below code to the lambda function:

import json
import urllib.parse
import boto3
import sys
import pandas as pd
import io
from elasticsearch import Elasticsearch

print('Loading function')

#s3 = boto3.client('s3')
s3 = boto3.resource('s3')

es = Elasticsearch("your_url")
index = "github_commit"
body = {
        "mappings": {
            "properties": {
                "CommitID": {
                    "type": "text"
                },
                "Name":{
                    "type": "text"
                },
                "Date": {
                    "type": "text"
                },
                "Address": {
                    "type": "text"
                }
            # }
            }
        },
        "settings": {
            "index" : {
                "number_of_shards" : 1,
                "number_of_replicas" : 0
            }
        }
        
}

def lambda_handler(event, context):
    #print("Received event: " + json.dumps(event, indent=2))

    # Get the object from the event and show its content type
    bucket = event['Records'][0]['s3']['bucket']['name']
    key = urllib.parse.unquote_plus(event['Records'][0]['s3']['object']['key'], encoding='utf-8')
    
    try:
    
        obj = s3.Object(bucket, key)
        s3data = obj.get()['Body'].read() #.decode('utf-8')
        csv = pd.read_excel(io.BytesIO(s3data))
        parsed_csv = csv.to_dict()
        print(parsed_csv)
        bulk_api_body = []
        print(len(parsed_csv['CommitID']))
        for i in range(0, len(parsed_csv['CommitID'])):
            action = {"index": {"_index": index, "_id": parsed_csv['CommitID'][i]}}
            eachcommit = {
                'CommitID': parsed_csv['CommitID'][i],
                ' Name': parsed_csv['Name'][i],
                'Date': parsed_csv['Date'][i],
                'Address': parsed_csv['Address'][i]
            }
            bulk_api_body.append(action)
            bulk_api_body.append(eachcommit)
        print(bulk_api_body)
        response = addData(bulk_api_body)
        if not response:
            print("could not add data to ES")
            return "could not add data to ES"
        
        return "Pushed to ES"
    except Exception as e:
        print(e)
        print('Error getting object {} from bucket {}. Make sure they exist and your bucket is in the same region as this function.'.format(key, bucket))
        raise e


def es_reachable():
    return es.ping()
    
def index_exist():
    return es.indices.exists(index=index)

def addData(bulk_api_body):
    if not es_reachable:
        print("ES not reachable")
        return false
    if not index_exist():
        print("Index Doesnot exist. Creating index")
        response = es.indices.create(index=index, body=body)  
        # settings=index_settings, mappings=index_mapping)
        if not response['acknowledged']:
            print("Failed to create index")
            return false
    print("adding data to ES")
    es.bulk(index=index, body=bulk_api_body)
    return True
        

   

Add trigger into the  lambda function:

So, to add trigger to lambda function use the below step:

• Select Function >> select Add trigger >>Trigger configuration >>  S3 Bucket >> Bucket (search for created bucket )  >> Add

 Also Create an IAM Role and attach AmazonS3FullAccess, CloudwatchFullAccess, AWSLambda_FullAccess policies to it:

Attach IAM Role to Lambda function during creating lambda under policies section:

Add Layers under AWS Lambda:

So, for create layers in Lambda Function:

For AWSSDKPandas-Python37

• Use AWS Lambda >> Layers >> Add Layer >> AWS Layers >> select (AWSSDKPandas-Python37) >> Version (3) >> Add 

For Elasticsearch layer upload the below zipfile

Zip File

• Search  AWS Lambda  >>  Create Layer  >>  Layer Configuration >> Name  >>  Upload a zip file  >>  compatible architecture (x86_64)  >> Compatible runtime (1)  >> Create 

Create an AWS S3 Bucket:

So, to create an s3 bucket follow the below steps:

• Open your AWS Console and search for service as S3
• Go to Amazon S3 >> Create Bucket >>  >> Mention Bucket name  >> Bucket Versioning Enabled  >> Create Bucket

Test the AWS Lambda Function by pushing data to S3 Bucket:

• Go to AWS S3 Bucket  >> Your-Bucket >> Click Upload >> Add files >> upload 

Once the data is pushed you’ll see that the AWS Lambda is triggered because of recent pushed logs to the s3 bucket . So, You can check this in the CloudWatch Logs as: 

So, this AWS Lambda Function will connect with the Elasticsearch and push the s3 bucket CSV File logs to es as:

So, now let’s Visualize the pushed logs in Kibana by creating the index-pattern to the specified indices and see the pushed logs as:

Finally, this is the whole process by using which we can push logs in CSV File format to Elasticsearch  through a lambda function.

Conclusion

So, I am done this is how to with this post in which we have seen how to create Function for pushing logs to ES. I hope you all will found this blog helpful and understandable to you because I have mentioned all the steps. Go and try it and let me know if you have any queries.

Thank You all !!!

Happy Learning 🙂

Reference

https://aws.plainenglish.io/ingest-log-streaming-data-into-elasticsearch-from-s3-b638d218d1f1

https://www.tutorialspoint.com/aws_lambda/aws_lambda_using_lambda_function_with_amazon_s3.htm