Skip to content

feat: Use a daemon thread to monitor the go feature server exclusively #2391

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
feast-ci-bot merged 23 commits into from
Mar 10, 2022
Merged

feat: Use a daemon thread to monitor the go feature server exclusively #2391

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
23 commits
Select commit Hold shift + click to select a range
612c962
feat: Use a daemon thread to monitor the go feature server exclusively
achals Mar 9, 2022
f14ac9b
fix workflow
achals Mar 9, 2022
78bbd66
Fix setup.py build_python_protos
achals Mar 9, 2022
8014662
CR comments
achals Mar 9, 2022
2ed95ec
mkdir is needed
achals Mar 9, 2022
bff6e92
explicit commit
achals Mar 9, 2022
7f33768
flush file explicitly
achals Mar 9, 2022
99fb066
Flush registry changes
achals Mar 9, 2022
88db3ec
Signal handler can only be triggered from main thread
achals Mar 9, 2022
31dbbf0
join on background thread when cancelling and update test
achals Mar 9, 2022
0f3afa3
fix redis
achals Mar 9, 2022
f5710ff
add back go_build.py
achals Mar 9, 2022
88e2c49
wait on grpc connection check
achals Mar 10, 2022
2fa1ccf
Add a more robust wait
achals Mar 10, 2022
92ff707
Add a sleep
achals Mar 10, 2022
54f1a08
Remove explicitt cleanup
achals Mar 10, 2022
67c7c17
More defensive
achals Mar 10, 2022
a79ec1c
Even more defensive
achals Mar 10, 2022
f8f0402
Stop join
achals Mar 10, 2022
635bd2a
Clean up the process to ensure next one is fine
achals Mar 10, 2022
3ea3fac
join thread
achals Mar 10, 2022
b83d4b3
remove a sleep
achals Mar 10, 2022
39a66e3
Cleanup
achals Mar 10, 2022
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
Prev Previous commit
Cleanup 
Signed-off-by: Achal Shah <achals@gmail.com>
  • Loading branch information
@achals
achals committed Mar 10, 2022
commit 39a66e3c7306a1d7ae09aefd0a49a60beb6de141
7 changes: 5 additions & 2 deletions sdk/python/feast/feature_store.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -729,6 +729,10 @@ def apply(
service.name, project=self.project, commit=False
)

# If a go server is running, kill it so that it can be recreated in `update_infra` with
# the latest registry state.
self.kill_go_server()

self._get_provider().update_infra(
project=self.project,
tables_to_delete=views_to_delete if not partial else [],
Expand All @@ -750,8 +754,7 @@ def teardown(self):

entities = self.list_entities()

if self._go_server:
self._go_server.kill_go_server_explicitly()
self.kill_go_server()

self._get_provider().teardown_infra(self.project, tables, entities)
self._registry.teardown()
Expand Down
17 changes: 10 additions & 7 deletions sdk/python/feast/go_server.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -246,28 +246,28 @@ def __init__(

def run(self):
# Target function of the thread class
_logger.info(
_logger.debug(
"%s Started monitoring thread to keep go feature server alive", self.ident
)
try:
while not self._is_cancelled.is_set():

# If we fail to connect to grpc stub, terminate subprocess and repeat
_logger.info("%s Connecting to subprocess", self.ident)
_logger.debug("%s Connecting to subprocess", self.ident)
if not self._shared_connection.connect():
_logger.info(
_logger.debug(
"%s Failed to connect, killing and retrying", self.ident
)
self._shared_connection.kill_process()
continue
else:
_logger.info(
_logger.debug(
"%s Go feature server started, process: %s",
self.ident,
self._shared_connection._process.pid,
)
self._go_server_started.set()
_logger.info(
_logger.debug(
"%s is_cancelled status: %s", self.ident, self._is_cancelled
)
while not self._is_cancelled.is_set():
Expand All @@ -276,7 +276,7 @@ def run(self):
self._shared_connection.wait_for_process(3600)
Comment thread
achals marked this conversation as resolved.
except subprocess.TimeoutExpired:
pass
_logger.info(
_logger.debug(
"%s No longer waiting for process: %s, %s, %s",
self.ident,
self._shared_connection._process.pid,
Expand All @@ -290,6 +290,9 @@ def run(self):
self._shared_connection.kill_process()

def stop(self):
# _logger.info("%s Stopping monitoring thread and terminating go feature server", self.ident)
_logger.debug(
"%s Stopping monitoring thread and terminating go feature server",
self.ident,
)
self._is_cancelled.set()
self._shared_connection.kill_process()