[Serve] Support manually terminating a replica

[dtran24]

Fixes #3135

Tested (run the relevant ones):

• Code formatting: bash format.sh
• Any manual or new tests for this PR (please specify below)
  • Terminated replicas with YAML specifying replicas: 2
• All smoke tests: pytest tests/test_smoke.py
• Relevant individual smoke tests: pytest tests/test_smoke.py::test_fill_in_the_name
• Backward compatibility tests: bash tests/backward_comaptibility_tests.sh

[dtran24]

Requesting review @cblmemo @MaoZiming

[cblmemo]

Thanks for adding this exciting feature!! This would be very helpful. Left some comments

[cblmemo]

It is not enough to only call replica_manager.scale_down since one important usecase of this command is cleanup failed records, but this function will keep any failed record. Should we add a purge argument to force clean it from database?

[dtran24]

Is something like this change (b838b1f) what you had in mind? If so, what's the best way to test this?

[cblmemo]

Yes it is, except for we need to terminate the cluster too when purging. IIUC current implementation will only remove the replica record. You can try to sky serve down --purge <service-name> -r <id> for : 1. a failed replica, 2. a ready replica and see if the record and cluster got removed.

[cblmemo]

Also, we should print some error message if we terminate a failed replica without --purge and prompt the user to use --purge to cleanup the record.

[cblmemo]

as said in the comment above, we should cleanup this cluster as well. how about adding an argument purge: bool in ReplicaManager.scale_down and remove the replica record here?

skypilot/sky/serve/replica_managers.py

Lines 752 to 775 in e51c2fb

	if info.status_property.is_scale_down_succeeded(
	self._get_initial_delay_seconds(info.version)):
	# This means the cluster is deleted due to
	# a scale down or the cluster is recovering
	# from preemption. Delete the replica info
	# so it won't count as a replica.
	if info.status_property.preempted:
	removal_reason = 'for preemption recovery'
	else:
	removal_reason = 'normally'
	# Don't keep failed record for version mismatch replicas,
	# since user should fixed the error before update.
	elif info.version != self.latest_version:
	removal_reason = 'for version outdated'
	else:
	logger.info(f'Termination of replica {replica_id} '
	'finished. Replica info is kept since some '
	'failure detected.')
	serve_state.add_or_update_replica(self._service_name,
	replica_id, info)
	if removal_reason is not None:
	serve_state.remove_replica(self._service_name, replica_id)
	logger.info(f'Replica {replica_id} removed from the '
	f'replica table {removal_reason}.')