데이터 레이크 쿼리

Panther API 검색 작업

개요

Panther API는 다음 데이터 레이크 작업을 지원합니다:

데이터 레이크 데이터베이스, 테이블, 열 나열
SQL을 사용하여 데이터 레이크(Data Explorer) 쿼리 실행
Search 쿼리 실행
현재 실행 중인 쿼리 취소
이전에 실행한 쿼리의 세부 정보 가져오기
현재 실행 중이거나 이전에 실행된 모든 쿼리를 선택적 필터와 함께 나열

Console의 API Playground 또는 GraphQL-over-HTTP API를 사용하여 Panther의 API를 호출할 수 있습니다. 이러한 방법에 대해 자세히 알아보려면 Panther API.

핵심 데이터 레이크 쿼리 작업에 관한 GraphQL 쿼리, mutation, 엔드투엔드 워크플로 예시는 아래 섹션을 참조하세요.

API를 통해 관리되는 쿼리는 SQL로 작성되어야 하며, 사용할 수 없습니다 PantherFlow.

일반적인 데이터 레이크 쿼리 작업

아래는 Panther에서 가장 흔한 GraphQL 데이터 레이크 쿼리 작업들입니다. 이 예시는 GraphQL 클라이언트(또는 curl)를 사용하여 Panther의 GraphQL API를 호출할 때 보내야 하는 문서를 보여줍니다.

데이터베이스 엔터티

# `AllDatabaseEntities`는 해당 작업의 별칭입니다
query AllDatabaseEntities {
  dataLakeDatabases {
     name
     description
     tables {
       name
       description
       columns {
         name
         description
         유형
       }
     }
   }
 }

# `DatabaseEntities`는 해당 작업의 별칭입니다
query DatabaseEntities {
  dataLakeDatabase(name: "panther_logs.public") {
     name
     description
     tables {
       name
       description
       columns {
         name
         description
         유형
       }
     }
  }
}

쿼리 실행

# `IssueDataLakeQuery`는 해당 작업의 별칭입니다
mutation IssueDataLakeQuery {
  executeDataLakeQuery(input: {
    sql: "select * from panther_logs.public.aws_alb limit 50"
  }) {
     id # 쿼리의 고유 ID
  }
}

# `IssueIndicatorSearchQuery`는 해당 작업의 별칭입니다
mutation IssueIndicatorSearchQuery {
  executeIndicatorSearchQuery(input: {
    indicators: ["286103014039", "126103014049"]
    startTime: "2022-04-01T00:00:00.000Z",
    endTime: "2022-04-30T23:59:59.000Z"
    indicatorName: p_any_aws_account_ids # 또는 자동 감지를 위해 비워 두세요
  }) {
     id # 쿼리의 고유 ID
  }
}

# `AbandonQuery`는 해당 작업의 별칭입니다
mutation AbandonQuery {
  cancelDataLakeQuery(input: { id: "1234-5678" }) {
     id # 취소된 ID를 반환
  }
}

데이터 레이크 또는 Search 쿼리의 결과 가져오기

데이터 레이크 또는 Search 쿼리를 실행하면 결과가 반환되는 데 몇 초에서 몇 분이 걸릴 수 있습니다. 쿼리가 완료되었는지 확인하려면 쿼리 상태를 확인해야 합니다(폴링).

다음 쿼리를 사용하여 쿼리 상태를 확인하면서, 가능한 경우 결과도 가져올 수 있습니다:

# `QueryResults`는 해당 작업의 별칭입니다
query QueryResults {
  dataLakeQuery(id: "1234-1234-1234-1234") { # 쿼리의 고유 ID
    message
    status
    results {
      edges {
        node
      }
    }
  }
}

# `QueryResults`는 해당 작업의 별칭입니다
query QueryResults {
  dataLakeQuery(id: "1234-1234-1234-1234") { # 쿼리의 고유 ID
    message
    status
    results(input: { cursor: "5678-5678-5678-5678" }) { # `endCursor`의 값
      edges {
        node
      }
      pageInfo
        endCursor
        hasNextPage
      }
    }
  }

다음의 예상 값은 status 및 results 쿼리 상태에 따라 달라집니다:

쿼리가 아직 실행 중이면:
- status 값은 running
- results 값은 null
쿼리가 실패했다면:
- status 값은 failed
- results 값은 null 그리고 오류 메시지는 다음에서 사용할 수 있습니다 message 키
쿼리가 완료되었다면
- status 값은 succeeded
- results 채워집니다

위의 모든 항목(다음의 가능한 값들과 함께 status) , 요청할 수 있는 추가 필드도 포함됩니다. 다음을 알아보세요 Panther API 스키마를 살펴보는 다양한 방법을 여기에서 확인하세요.

데이터 레이크 또는 Search 쿼리의 메타데이터 가져오기

위의 예제에서는 Panther 쿼리의 results 를 요청했습니다. 쿼리에 관한 추가 메타데이터를 요청하는 것도 가능합니다.

다음 예제에서는 결과의 첫 페이지와 함께 이러한 메타데이터를 요청합니다:

# `QueryMetadata`는 해당 작업의 별칭입니다
query QueryMetadata {
  dataLakeQuery(id: "1234-1234-1234-1234") { # 쿼리의 고유 ID
    name
    isScheduled
    issuedBy {
      ... on User {
        email
      }
      ... on APIToken {
        name
      } 
    }
    sql
    message
    status
    startedAt
    completedAt
    results {
      edges {
        node
      }
    }
  }
}

데이터 레이크 및 Search 쿼리 나열

# `ListDataLakeQueries`는 해당 작업의 별칭입니다
query ListDataLakeQueries {
  dataLakeQueries {
    name
    isScheduled
    issuedBy {
      ... on User {
        email
      }
      ... on APIToken {
        name
    } 
    }
    sql
    message
    status
    startedAt
    completedAt
    results { # 각 쿼리의 결과 첫 페이지만 가져옵니다
      edges {
        node
      }
    }
  }

# `ListDataLakeQueries`는 해당 작업의 별칭입니다
query ListDataLakeQueries {
  dataLakeQueries(input: { cursor: "5678-5678-5678-5678" }) { # `endCursor`의 값
    name
    isScheduled
    issuedBy {
      ... on User {
        email
      }
      ... on APIToken {
        name
    } 
    }
    sql
    message
    status
    startedAt
    completedAt
    results { # 각 쿼리의 결과 첫 페이지만 가져옵니다
      edges {
        node
      }
    }
    pageInfo {
      endCursor
      hasNextPage
    }
  }

# `ListDataLakeQueries`는 해당 작업의 별칭입니다
query ListDataLakeQueries {
  dataLakeQueries(input: { contains: "aws_alb", isScheduled: true }) {
    name
    isScheduled
    issuedBy {
      ... on User {
        email
      }
      ... on APIToken {
        name
    } 
    }
    sql
    message
    status
    startedAt
    completedAt
    results { # 각 쿼리의 결과 첫 페이지만 가져옵니다
      edges {
        node
      }
    }
  }

엔드투엔드 예시

아래에서는 다음을 기반으로 설명하겠습니다: 일반 작업 엔드투엔드 흐름을 보여주는 예시.

데이터 레이크(Data Explorer) 쿼리 실행

// npm install graphql graphql-request

import { GraphQLClient, gql } from 'graphql-request';

const client = new GraphQLClient(
  'YOUR_PANTHER_API_URL', 
  { headers: { 'X-API-Key': 'YOUR_API_KEY' } 
});

// `IssueQuery`는 쿼리의 별칭입니다. 완전히 생략할 수 있습니다.
const issueQuery = gql`
  mutation IssueQuery($sql: String!) {
    executeDataLakeQuery(input: { sql: $sql }) {
      id
    }
  }
`;

// `GetQueryResults`는 쿼리의 별칭입니다. 완전히 생략할 수 있습니다.
const getQueryResults = gql`
  query GetQueryResults($id: ID!, $cursor: String) {
    dataLakeQuery(id: $id) {
      message
      status
      results(input: { cursor: $cursor }) {
        edges {
          node
        }
        pageInfo {
          endCursor
          hasNextPage
        }
      }
    }
  }
`;

(async () => {
  try {
    // 가져오는 모든 결과 노드를 보관하는 누산기
    let allResults = [];
    // 루프를 언제 종료할지 알기 위한 헬퍼
    let hasMore = true;
    // 페이지네이션 커서
    let cursor = null;

    // 쿼리 실행
    const mutationData = await client.request(issueQuery, {
      sql: 'select * from panther_logs.public.aws_alb limit 5',
    });

    // 쿼리가 결과를 반환할 때까지 폴링을 시작합니다. 그다음,
    // 더 이상 남은 페이지가 없을 때까지 계속 가져옵니다
    do {
      const queryData = await client.request(getQueryResults, {
        id: mutationData.executeDataLakeQuery.id,
        cursor,
      });

      // 아직 실행 중이면 메시지를 출력하고 폴링을 계속합니다
      if (queryData.dataLakeQuery.status === 'running') {
        console.log(queryData.dataLakeQuery.message);
        continue;
      }

      // 실행 중도 아니고 완료도 아니라면,
      // 취소되었거나 오류가 발생한 것입니다. 이 경우,
      // 예외를 발생시킵니다
      if (queryData.dataLakeQuery.status !== 'succeeded') {
        throw new Error(queryData.dataLakeQuery.message);
      }

      allResults = [...allResults, ...queryData.dataLakeQuery.results.edges.map(edge => edge.node)ាស់];

      hasMore = queryData.dataLakeQuery.results.pageInfo.hasNextPage;
      cursor = queryData.dataLakeQuery.results.pageInfo.endCursor;
    } while (hasMore);

    console.log(`쿼리에서 ${allResults.length}개의 결과가 반환되었습니다!`);
  } catch (err) {
    console.error(err.response);
  }
})();

# pip install gql aiohttp

from gql import gql, Client
from gql.transport.aiohttp import AIOHTTPTransport

transport = AIOHTTPTransport(
  url="YOUR_PANTHER_API_URL",
  headers={"X-API-Key": "YOUR_API_KEY"}
)

client = Client(transport=transport, fetch_schema_from_transport=True)

# `IssueQuery`는 쿼리의 별칭입니다. 완전히 생략할 수 있습니다.
issue_query = gql(
    """
    mutation IssueQuery($sql: String!) {
        executeDataLakeQuery(input: { sql: $sql }) {
            id
        }
    }
    """
)

# `GetQueryResults`는 쿼리의 별칭입니다. 완전히 생략할 수 있습니다.
get_query_results = gql(
    """
    query GetQueryResults($id: ID!, $cursor: String) {
        dataLakeQuery(id: $id) {
            message
            status
            results(input: { cursor: $cursor }) {
                edges {
                    node
                }
                pageInfo {
                    endCursor
                    hasNextPage
                }
            }
        }
    }
    """
)

# 모든 페이지에서 가져오는 모든 결과를 보관하는 누산기
all_results = []
# 루프를 언제 종료할지 알기 위한 도우미.
has_more = True
# 페이지네이션 커서
cursor = None

# 데이터 레이크(Data Explorer) 쿼리 실행
mutation_data = client.execute(
    issue_query,
    variable_values={
        "sql": "select * from panther_logs.public.aws_alb limit 5"
    }
)

# 쿼리가 결과를 반환할 때까지 폴링을 시작합니다. 그다음,
# 더 이상 남은 페이지가 없을 때까지 계속 가져옵니다
while has_more:
    query_data = client.execute(
        get_query_results,
        variable_values = {
            "id": mutation_data["executeDataLakeQuery"]["id"],
            "cursor": cursor
        }
    )
    
    # 아직 실행 중이면 메시지를 출력하고 폴링을 계속합니다
    if query_data["dataLakeQuery"]["status"] == "running":
        print(query_data["dataLakeQuery"]["message"])
        continue
    
    # 실행 중도 아니고 완료도 아니라면,
    # 취소되었거나 오류가 발생한 것입니다. 이 경우,
    # 예외를 발생시킵니다
    if query_data["dataLakeQuery"]["status"] != "succeeded":
        raise Exception(query_data["dataLakeQuery"]["message"])

    all_results.extend([edge["node"] for edge in query_data["dataLakeQuery"]["results"]["edges"]])
    has_more = query_data["dataLakeQuery"]["results"]["pageInfo"]["hasNextPage"]
    cursor = query_data["dataLakeQuery"]["results"]["pageInfo"]["endCursor"]

print(f'쿼리에서 {len(all_results)}개의 결과가 반환되었습니다!')

Search 쿼리 실행

// npm install graphql graphql-request

import { GraphQLClient, gql } from 'graphql-request';

const client = new GraphQLClient(
  'YOUR_PANTHER_API_URL', 
  { headers: { 'X-API-Key': 'YOUR_API_KEY' } 
});

// `IssueQuery`는 쿼리의 별칭입니다. 완전히 생략할 수 있습니다.
const issueQuery = gql`
  mutation IssueQuery($input: ExecuteIndicatorSearchQueryInput!) {
    executeIndicatorSearchQuery(input: $input) {
      id
    }
  }
`;

// `GetQueryResults`는 쿼리의 별칭입니다. 완전히 생략할 수 있습니다.
const getQueryResults = gql`
  query GetQueryResults($id: ID!, $cursor: String) {
    dataLakeQuery(id: $id) {
      message
      status
      results(input: { cursor: $cursor }) {
        edges {
          node
        }
        pageInfo {
          endCursor
          hasNextPage
        }
      }
    }
  }
`;

(async () => {
  try {
    // 가져오는 모든 결과 노드를 보관하는 누산기
    let allResults = [];
    // 루프를 언제 종료할지 알기 위한 헬퍼
    let hasMore = true;
    // 페이지네이션 커서
    let cursor = null;

    // 쿼리 실행
    const mutationData = await client.request(issueQuery, {
      input: {         
        indicators: ["226103014039"],
        startTime: "2022-03-29T00:00:00.001Z",
        endTime: "2022-03-30T00:00:00.001Z",
        indicatorName: "p_any_aws_account_ids"
      }
    });

    // 더 이상 남은 페이지가 없을 때까지 계속 가져오기
    do {
      const queryData = await client.request(getQueryResults, {
        id: mutationData.executeIndicatorSearchQuery.id,
        cursor,
      });

      // 아직 실행 중이면 메시지를 출력하고 폴링을 계속합니다
      if (queryData.dataLakeQuery.status === 'running') {
        console.log(queryData.dataLakeQuery.message);
        continue;
      }

      // 실행 중도 아니고 완료도 아니라면,
      // 취소되었거나 오류가 발생한 것입니다. 이 경우,
      // 예외를 발생시킵니다
      if (queryData.dataLakeQuery.status !== 'succeeded') {
        throw new Error(queryData.dataLakeQuery.message);
      }

      allResults = [...allResults, ...queryData.dataLakeQuery.results.edges.map(edge => edge.node)ាស់];

      hasMore = queryData.dataLakeQuery.results.pageInfo.hasNextPage;
      cursor = queryData.dataLakeQuery.results.pageInfo.endCursor;
    } while (hasMore);

    console.log(`쿼리에서 ${allResults.length}개의 결과가 반환되었습니다!`);
  } catch (err) {
    console.error(err.response);
  }
})();

# pip install gql aiohttp

from gql import gql, Client
from gql.transport.aiohttp import AIOHTTPTransport

transport = AIOHTTPTransport(
  url="YOUR_PANTHER_API_URL",
  headers={"X-API-Key": "YOUR_API_KEY"}
)

client = Client(transport=transport, fetch_schema_from_transport=True)

# `IssueQuery`는 쿼리의 별칭입니다. 완전히 생략할 수 있습니다.
issue_query = gql(
    """
    mutation IssueQuery($input: ExecuteIndicatorSearchQueryInput!) {
        executeIndicatorSearchQuery(input: $input) {
            id
        }
    }
    """
)

# `GetQueryResults`는 쿼리의 별칭입니다. 완전히 생략할 수 있습니다.
get_query_results = gql(
    """
    query GetQueryResults($id: ID!, $cursor: String) {
        dataLakeQuery(id: $id) {
            message
            status
            results(input: { cursor: $cursor }) {
                edges {
                    node
                }
                pageInfo {
                    endCursor
                    hasNextPage
                }
            }
        }
    }
    """
)

# 모든 페이지에서 가져오는 모든 결과를 보관하는 누산기
all_results = []
# 루프를 언제 종료할지 알기 위한 헬퍼
has_more = True
# 페이지네이션 커서
cursor = None

# Indicator Search 쿼리 실행
mutation_data = client.execute(
    issue_query,
    variable_values={
        "input": {         
            "indicators": ["226103014039"],
            "startTime": "2022-03-29T00:00:00.001Z",
            "endTime": "2022-03-30T00:00:00.001Z",
            "indicatorName": "p_any_aws_account_ids"
        }
    }
)

# 쿼리가 결과를 반환할 때까지 폴링을 시작합니다. 그다음,
# 더 이상 남은 페이지가 없을 때까지 계속 가져옵니다
while has_more:    
    query_data = client.execute(
        get_query_results,
        variable_values = {
            "id": mutation_data["executeIndicatorSearchQuery"]["id",
            "cursor": cursor
        }
    )
    
    # 아직 실행 중이면 메시지를 출력하고 폴링을 계속합니다
    if query_data["dataLakeQuery"]["status"] == "running":
        print(query_data["dataLakeQuery"]["message"])
        continue
    
    # 실행 중도 아니고 완료도 아니라면,
    # 취소되었거나 오류가 발생한 것입니다. 이 경우,
    # 예외를 발생시킵니다
    if query_data["dataLakeQuery"]["status"] != "succeeded":
        raise Exception(query_data["dataLakeQuery"]["message"])

    all_results.extend([edge["node"] for edge in query_data["dataLakeQuery"]["results"]["edges"]])
    has_more = query_data["dataLakeQuery"]["results"]["pageInfo"]["hasNextPage"]
    cursor = query_data["dataLakeQuery"]["results"]["pageInfo"]["endCursor"]

print(f'쿼리에서 {len(all_results)}개의 결과가 반환되었습니다!')

이전클라우드 계정 관리 다음로그 소스 관리

마지막 업데이트 1년 전

도움이 되었나요?