I took some time this past weekend to play around with automating GoldenGate with Python and the REST APIs. Below, find the source code of a #Python script that gives you some extract and replicat statuses in a given deployment. I look forward to enhancing this with multiple deployments.

I wish everyone that has already made a move to GoldenGate Microservices would look at this and find the tremendous opportunity to improve your day-to-day operations. This is just an example of checking the extract and replicat status, and there are tons of other metrics you could bring over. To me (and again, that’s my personal opinion), this is the most powerful feature the platform provides, and still in its earlier days.

I hope this example will spark some curiosity and creativity. I would love to hear about your experiments and if you develop something really cool, let me know. Think about sending these metrics to other platforms like InfluxDB, Grafana, OpenTelemetry, Prometheus, etc.

In this example, I run the program on my own laptop with GoldenGate deployed locally on a docker image and the databases hosted in OCI.

The open source code below is for education and training purposes only, not supported by myself or Oracle. Use at your own risk, and I highly recommend fully testing and modifying as needed to fulfill your needs.

The code will first check for the number of extract and replicats and will loop through each process and display the metrics I selected. You can add more metrics, find in the REST documentation here: https://docs.oracle.com/en/middleware/goldengate/core/21.3/oggra/

One of the major advantages is that you can run/execute this from pretty much anywhere, On-Prem, on Cloud, local laptop, on 3rd party server, and even on mobile if you really want to get fancy.

The requirement for this to run is Python3 installed and make sure to adjust the endpoint variable described in the Config.

################################################################
# GoldenGate HUB Check Extract and Replicats
#
# Author: Alex Lima
# 
# Execution: $python3 check_gg_processes.py
#
# Config: Set GoldenGate HUR URL and PORT:  goldengate_hub_url
#
################################################################


import requests
import json
import os

# Clearing the Screen
os.system('clear')

# Disable Warning for InsecureRequestWarning certificate verification
requests.packages.urllib3.disable_warnings()

# Url and PORT for GOldengate HUB
goldengate_hub_url="https://localhost:415"

# DIsplay the Header content
def display_header():
 print("---------------------------------------------------------")
 print("-- GoldenGate HUB Status for " + goldengate_hub_url)
 print("---------------------------------------------------------")

# Print Extract and Replicat Processes Status
def print_proc_status(proc_name, hub_url, proc_type):
 url = hub_url + "/" + proc_name
 url_info = hub_url + "/" + proc_name + "/info/status"
 #print (url + "     " + url_info)

 response = requests.get(url, verify=False, headers=header)
 response_info = requests.get(url_info, verify=False, headers=header)

 #If full response display is required
 #print(json.dumps(response.json(), indent=4))
 #print(json.dumps(response_info.json(), indent=4))

 # Formated Response
 ##  Config for the Extract Formating
 if (proc_type == "extract"):
## Config for the Extract Formating
  rdata=response.json()
  rstatus = rdata['response']['status']
  rfilename = rdata['response']['targets'][0]['name']
  rsequence = rdata['response']['targets'][0]['sequence']
  roffset = rdata['response']['targets'][0]['offset']
  
  rdata_info=response_info.json()
  rlag = rdata_info['response']['lag']
  rlagSinceCheckpoint = rdata_info['response']['sinceLagReported']
 else:
## Config for the Replicat Formating
  rdata_info=response_info.json()
  rstatus = rdata_info['response']['status']
  rfilename = rdata_info['response']['position']['name']
  rsequence = rdata_info['response']['position']['sequence']
  roffset = rdata_info['response']['position']['offset']
  rlag = rdata_info['response']['lag']
  rlagSinceCheckpoint = rdata_info['response']['sinceLagReported']

## Print the Process Status
 print()
 print("Process Type: " + proc_type.upper() + "\nProcess Name: " + proc_name + "\nStatus: %s\nSequence: %s\nRBA: %s\nLag in Secs: %s\nTime Since Last Checkpoint: %s\nFile Name: %s"
  %(rstatus,rsequence,roffset,rlag,rlagSinceCheckpoint,rfilename))
 print()

# Check how many processes exist and loop through each one for status reporting
def check_ggprocess(ggprocess):
  if (ggprocess == "extract"):
   hub_url = goldengate_hub_url +"/services/v2/extracts"
  else:
   hub_url = goldengate_hub_url +"/services/v2/replicats"

  response = requests.get(hub_url, verify=False, headers=header).text
  response_info = json.loads(response)

  # Loop along dictionary keys
  i = 0
  while i < len(response_info['response']['items']):
    print_proc_status(response_info['response']['items'][i]['name'], hub_url, ggprocess)
    i += 1

# Authentication
header = {"Authorization" : "Basic b2dnYWRtaW46V3dlbGNvbUUjIzEyMw=="}
##---

# Main
display_header()
check_ggprocess("extract")
check_ggprocess("replicat")

You can use this free tool to get your Authorization code: https://reqbin.com/post-online