Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
221 changes: 221 additions & 0 deletions src/impl/internal_node_utils.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,221 @@
from __future__ import annotations

from . import internal_utils
from . platforms.internal_platform_utils_provider import InternalPlaformUtilsProvider

from .. import consts
from ..node_state import PostgresNodeState
from ..enums import NodeStatus
from ..raise_error import RaiseError
from ..exceptions import ExecUtilException

from testgres.operations.os_ops import OsOperations

import typing
import time


class InternalNodeUtils:
T_PLATFORM_UTILS = InternalPlaformUtilsProvider.T_PLATFORM_UTILS

@staticmethod
def get_pg_node_state(
os_ops: OsOperations,
bin_dir: str,
data_dir: str,
utils_log_file: typing.Optional[str],
) -> PostgresNodeState:
assert isinstance(os_ops, OsOperations)
assert type(bin_dir) is str
assert type(data_dir) is str
assert utils_log_file is None or type(utils_log_file) is str

C_MAX_ATTEMPTS = 3
C_SLEEP_TIME1 = 1
C_SLEEP_TIME_MULT = 2

pg_ctl_params = [
os_ops.build_path(bin_dir, consts.BINARY_NAME__PG_CTL),
"-D",
data_dir,
"status",
]

attempt = 0
sleep_time = C_SLEEP_TIME1

platform_utils_provider = InternalPlaformUtilsProvider(
os_ops,
)

while True:
assert type(attempt) is int
assert attempt >= 0
assert attempt < C_MAX_ATTEMPTS

attempt += 1

if attempt > 1:
internal_utils.send_log_debug("Sleep {} second(s) before an attempt #{}".format(
sleep_time,
attempt,
))
time.sleep(sleep_time)
sleep_time = sleep_time * C_SLEEP_TIME_MULT

exec_r = internal_utils.execute_utility3(
os_ops,
pg_ctl_params,
utils_log_file,
check=False,
)

status_code = exec_r.returncode
out = exec_r.stdout
error = exec_r.stderr

assert type(status_code) is int
assert type(out) is str
assert type(error) is str

# -----------------
if status_code == consts.PG_CTL__STATUS__NODE_IS_STOPPED:
return PostgresNodeState(NodeStatus.Stopped, None)

# -----------------
if status_code == consts.PG_CTL__STATUS__BAD_DATADIR:
return PostgresNodeState(NodeStatus.Uninitialized, None)

# -----------------
if status_code == consts.PG_CTL__STATUS__OK:
pid = __class__._parse_pid(
out,
pg_ctl_params,
)
assert type(pid) is int
assert pid != 0

# ----------------- detect zombie
if platform_utils_provider.get().ProcessIsZombie_soft_check(os_ops, pid) is True:
internal_utils.send_log_debug("Postmaster process {} is a zombie.".format(
pid,
))
return PostgresNodeState(NodeStatus.Zombie, pid)

# -----------------
return PostgresNodeState(NodeStatus.Running, pid)

assert status_code != consts.PG_CTL__STATUS__OK

errMsg = "Getting of a node status [data_dir is {0}] failed.".format(
data_dir,
)

e1 = ExecUtilException(
message=errMsg,
command=pg_ctl_params,
exit_code=status_code,
out=out,
error=error,
)

if status_code == consts.PG_CTL__STATUS__FAILED:
internal_utils.send_log_debug(
"pg_ctl fails with an error: {}".format(
exec_r.stderr,
))

try:
find_postmaster_r = platform_utils_provider.get().FindPostmaster(
os_ops,
bin_dir,
data_dir,
)
except Exception as e2:
raise e2 from e1

assert type(find_postmaster_r) is __class__.T_PLATFORM_UTILS.FindPostmasterResult

if find_postmaster_r.code == __class__.T_PLATFORM_UTILS.FindPostmasterResultCode.ok:
# Postmaster is alive. Let's wait a few seconds and check its status again.
internal_utils.send_log_debug(
"Postmaster is found and has PID {}.".format(
find_postmaster_r.pid,
))

if attempt < C_MAX_ATTEMPTS:
continue

raise e1

@staticmethod
def _parse_pid(
out: str,
pg_ctl_params,
) -> int:
assert type(out) is str

if out == "":
RaiseError.pg_ctl_returns_an_empty_string(
pg_ctl_params,
)

C_PID_PREFIX = "(PID: "

i = out.find(C_PID_PREFIX)

if i == -1:
RaiseError.pg_ctl_returns_an_unexpected_string(
out,
pg_ctl_params,
)

assert i > 0
assert i < len(out)
assert len(C_PID_PREFIX) <= len(out)
assert i <= len(out) - len(C_PID_PREFIX)

i += len(C_PID_PREFIX)
start_pid_s = i

while True:
if i == len(out):
RaiseError.pg_ctl_returns_an_unexpected_string(
out,
pg_ctl_params,
)

ch = out[i]

if ch == ")":
break

if ch.isdigit():
i += 1
continue

RaiseError.pg_ctl_returns_an_unexpected_string(
out,
pg_ctl_params,
)
assert False

if i == start_pid_s:
RaiseError.pg_ctl_returns_an_unexpected_string(
out,
pg_ctl_params,
)

# TODO: Let's verify a length of pid string.

pid = int(out[start_pid_s:i])

if pid == 0:
RaiseError.pg_ctl_returns_a_zero_pid(
out,
pg_ctl_params,
)

assert pid != 0

return pid
58 changes: 58 additions & 0 deletions src/impl/internal_utils.py
Original file line number Diff line number Diff line change
@@ -1,4 +1,10 @@
from testgres.operations.os_ops import OsOperations
from testgres.operations.os_ops import OsCommandResult
from testgres.operations.types import T_OS_CMD
from testgres.operations.types import T_OS_EXEC_ENV
from testgres.operations.helpers import Helpers as OsHelpers

from ..exceptions import ExecUtilException

import logging
import typing
Expand Down Expand Up @@ -90,3 +96,55 @@ def read_line_to_pos__bin(
assert type(result) is bytes
assert len(result) <= (position - read_position)
return result


def execute_utility3(
os_ops: OsOperations,
args: T_OS_CMD,
logfile: typing.Optional[str] = None,
check: bool = True,
exec_env: typing.Optional[T_OS_EXEC_ENV] = None,
) -> OsCommandResult:
assert os_ops is not None
assert isinstance(os_ops, OsOperations)
assert type(check) is bool
assert exec_env is None or type(exec_env) is dict

exec_r = os_ops.run(
args,
check=check,
encoding=OsHelpers.GetDefaultEncoding(),
exec_env=exec_env,
)

assert type(exec_r) is OsCommandResult

# write new log entry if possible
if logfile:
try:
log_lines = [
os_ops.join_command_arguments(args),
]

if exec_r.stdout is None:
log_lines.append("# #NONE#")
else:
# comment-out lines
assert type(exec_r.stdout) is str
log_lines += ['# ' + line for line in exec_r.stdout.splitlines()]

log_lines.append("")

os_ops.write(
filename=logfile,
data="\n".join(log_lines),
truncate=False,
)
except IOError:
raise ExecUtilException(
"Problem with writing to logfile `{}` during run command `{}`".format(
logfile,
args,
))

return exec_r
32 changes: 32 additions & 0 deletions src/impl/platforms/internal_platform_utils_provider.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,32 @@
from .internal_platform_utils_factory import create_internal_platform_utils
from .internal_platform_utils_factory import InternalPlatformUtils

from testgres.operations.os_ops import OsOperations

import typing


class InternalPlaformUtilsProvider:
T_PLATFORM_UTILS = InternalPlatformUtils

_os_ops: OsOperations
_platform_utils: typing.Optional[T_PLATFORM_UTILS] = None

def __init__(
self,
os_ops: OsOperations,
):
assert isinstance(os_ops, OsOperations)
self._os_ops = os_ops
self._platform_utils = None
return

def get(self) -> T_PLATFORM_UTILS:
if self._platform_utils is None:
self._platform_utils = create_internal_platform_utils(
self._os_ops,
)
assert isinstance(self._platform_utils, __class__.T_PLATFORM_UTILS)

assert isinstance(self._platform_utils, __class__.T_PLATFORM_UTILS)
return self._platform_utils
20 changes: 20 additions & 0 deletions src/node_state.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,20 @@
from .enums import NodeStatus

import typing


class PostgresNodeState:
node_status: NodeStatus
pid: typing.Optional[int]

def __init__(
self,
node_status: NodeStatus,
pid: typing.Optional[int],
):
assert type(node_status) is NodeStatus
assert pid is None or type(pid) is int

self.node_status = node_status
self.pid = pid
return
Loading
Loading