Test Rest API through python

Elastic Stack Elasticsearch
1

We have Elastic search cluster with thousands of applications. We would like to read logs through python script from elastic cluster.

Use Cases :

  1. Get logs of last 1 hour or 1 day
  2. If it has error 500 or 404

We would like to get this kind of details from cluster.

Able to connect to cluster and in body provided query with match condition.

But I was struck to provide filter for errors from logs and getting last 1h data.

2

Thanks for reaching out, @navya_k. Do you have an example of the script you running? Also what version of Elastic are you using?

3

Hello
Thank you for reply. We using 8.8.1 version.
And below is the code snippet:

1000055481
10000554811080×2400 76.6 KB

4

Thanks for providing this sample. It was a bit hard to read in a picture but it looks like you are testing with a sample flask app. My coworker @iulia wrote up a simple guide on this that might be helpful here.

5

Thank you so much for your reply and I will check code & run to see result

6

Hello,

I have tried the code. It has thrown error

ModuleNotFoundError: No module named 'streamlit'.

Tried to resolve with
Pip install streamlit

But its failed with error:
Preparing metadata (project.toml)...error
Exit code: 1
The meson build system
Vers : 1.2.1
Build type: native build
Project name: pandas
Project version: 2.2.2.
.
.
Run-time dependency python found : No ( tried sysconfig)
....\pandas_libs\tslibs\meson.build:32:7: Error : Python dependency not found.

How to resolve it ?

7

Thanks for following up. You can try this:

pip install elasticsearch flask
8

Hello,

Still..no luck.

Two points here

  1. Ran the command : pip install elasticsearch flask

After completion of pip....I ran the web_app.py
But still facing error

ModuleNotFoundError: No module named ' streamlit'.

  1. I have commented line

import streamlit as st

I can able to run - run module in python IDE. It has shown as
= RESTART: C:\users\Documents\web_app.py====

In browser.. 127.0.0.1:5000/search
It says
Can't reach the page

9

Our elastic cluster is on- primses, I have provide Elasticsearch ('http:/url', 'API_key')

Is that correct ?

10

Thanks for following up; you may want to try something like this:


pip install elasticsearch flask streamlit

In terms of connecting to Elasticsearch, this guide could be helpful. Let me know if you need any specific help here. This getting started guide may also be good to explore.

11

Please don't post images of text as they are hard to read, may not display correctly for everyone, and are not searchable.

Instead, paste the text and format it with </> icon or pairs of triple backticks (```), and check the preview window to make sure it's properly formatted before posting it. This makes it more likely that your question will receive a useful answer.

1 Like
12

Thank you for reply jessgarson.

It's not working for me..as my system not compatable with flask.

I have raised another discussion. Here is the link :

Please have a look and help me out for solution...

13

Sure. Thank you for approval.

14

Thanks for following up here, @navya_k. As @dadoonet mentioned, it's easier to keep everything in one thread. I and others from Elastic usually block out some time daily to answer questions, but sometimes, there is some back-and-forth to find a solution.

I had a chance to look at the code you provided and was wondering if something like this would work:

from elasticsearch import Elasticsearch

# Initialize the Elasticsearch client
es = Elasticsearch("https://localhost:9200", api_key="example_api_key")

# Define the index pattern
index_pattern = "test-*"

# Input for the service name
service_name = input("Enter service name: ")

# Define the search query
query = {
    "query": {
        "bool": {
            "must": [
                {
                    "range": {
                        "@timestamp": {
                            "gte": "now-1h"
                        }
                    }
                },
                {
                    "term": {
                        "service.name": service_name
                    }
                }
            ]
        }
    }
}

# Execute the search query
result = es.search(index=index_pattern, body=query)

# Print each hit's source data
print("Search Results:")
for hit in result['hits']['hits']:  # Correct the path to hits
    print(hit['_source'])

I also had a few follow up questions:

  • What is the role of service_name in your query?
  • What errors are you getting, if any?
  • How are you hosting Elasticsearch?

I also wrote a simple script that takes in some sample logs and uploads them to Elasticsearch:

from elasticsearch import Elasticsearch
from getpass import getpass

# Initialize Elasticsearch client
client = Elasticsearch(
    getpass("Host: "),
    api_key=getpass("Elastic API Key: "),
)

# Uploads logs to a specified Elasticsearch index.
def upload_logs(index_name, logs):
    for log in logs:
        client.index(index=index_name, document=log)

#  Searches logs in a specified Elasticsearch index.
def search_logs(index_name, query):
    return client.search(index=index_name, query={"match": query})

# Example usage:
if __name__ == "__main__":
    # Define the index name
    index_name = "sample_logs"

    # Example log entries
    logs = [
        {"timestamp": "2024-04-23T12:00:00", "level": "INFO", "message": "System start."},
        {"timestamp": "2024-04-23T12:05:00", "level": "ERROR", "message": "Failed to connect to database."},
    ]

    # Upload logs
    upload_logs(index_name, logs)

    # Search logs
    search_results = search_logs(index_name, {"message": "database"})
    print(search_results)

Thanks again! My coworker @iulia and I also have a repository of Python scripts we use regularly, which could be a helpful resource here.

15

Thank you for your reply.

  1. The role of service_ name it has.. all admin roles. Service name means which we can see under observability - APM - services.

  2. I am not seeing errors..the variable value not passing to query so it showing any result.

  3. Elastic it's a cluster on-prim cloud.

16

To debug effectively, you need to proceed in steps and answer all those questions:

  1. Do you have evidence that your query with a given service name works in Kibana Dev Tools Console?
  2. What is the exact code that you're running?
  3. What is the exact output you get?
  4. What do you expect instead?
1 Like
17

Hello,

I have tried the service name variable in query and got the result like below:

Service_ name: input("Enter name :")
query = "query": {
"match": {
"service.name": service_name

Output:
Enter name : 1ABC_Test
'Id': 1, type: filebeat, version: 8.12.0, message: 1ABC_chat-data-service, transaction: GetContactDetails, host: cruvify246, application name: 1ABC.
.
.
.

Here actually I am expecting data from only transactions. Name like transaction names.

Could you please help ??

18

Thank you jessgarson for reply. Your code is very helpfull for me.

One question here :

In your code logs you provided 5 mins time like below:
Logs = [
{"timestamp": " 2024-04-23T12:00:00", "message": "ElasticAPM"},
{" timestamp":"2024-04-23T12:05:00", "level": "ERROR"}

So..my question here is..instead provide time in code..how can we ask user to provide time as input ?

I am looking in python not in flask. Could you able to assist ?

19

Thanks, @navya_k, I'm glad you are starting to make some headway here.

You can adjust the search_logs function to take in user input and format it.

def search_logs(index_name):
    # Get user input for start and end time
    start_time = input("Enter start time (YYYY-MM-DDTHH:MM:SS): ")
    end_time = input("Enter end time (YYYY-MM-DDTHH:MM:SS): ")

    # Convert the input times to ISO format
    start_time_iso = datetime.strptime(start_time, '%Y-%m-%dT%H:%M:%S').isoformat()
    end_time_iso = datetime.strptime(end_time, '%Y-%m-%dT%H:%M:%S').isoformat()

    # Construct the query
    query = {
        "range": {
            "timestamp": {
                "gte": start_time_iso,
                "lte": end_time_iso
            }
        }
    }

    # Execute the search query
    response = client.search(index=index_name, body={"query": query})
    return response['hits']['hits']

The full code should look something like this:

from elasticsearch import Elasticsearch
from getpass import getpass
from datetime import datetime

# Initialize Elasticsearch client
client = Elasticsearch(
    getpass("Host: "),
    api_key=getpass("Elastic API Key: "),
)

# Uploads logs to a specified Elasticsearch index.
def upload_logs(index_name, logs):
    for log in logs:
        client.index(index=index_name, document=log)

# Searches logs in a specified Elasticsearch index based on a time range provided by the user.
def search_logs(index_name):
    # Get user input for start and end time
    start_time = input("Enter start time (YYYY-MM-DDTHH:MM:SS): ")
    end_time = input("Enter end time (YYYY-MM-DDTHH:MM:SS): ")

    # Convert the input times to ISO format
    start_time_iso = datetime.strptime(start_time, '%Y-%m-%dT%H:%M:%S').isoformat()
    end_time_iso = datetime.strptime(end_time, '%Y-%m-%dT%H:%M:%S').isoformat()

    # Construct the query
    query = {
        "range": {
            "timestamp": {
                "gte": start_time_iso,
                "lte": end_time_iso
            }
        }
    }

    # Execute the search query
    response = client.search(index=index_name, body={"query": query})
    return response['hits']['hits']

# Example usage:
if __name__ == "__main__":
    # Define the index name
    index_name = "sample_logs"

    # Upload logs
    upload_logs(index_name, logs)

    # Search logs within a time range specified by the user
    search_results = search_logs(index_name)
    print(search_results)
20

Thank you so much for you inputs and code.

I have executed it.. but it showing as output...not showing logs

Output :
Enter start time : 2024-04-29T03:30:00
Enter end time: 2024-04-29T04:30:00

And getting out of output screen.

Not sure why it's not printing the logs as result