1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
|
# Copyright (C) 2017-2022 Vanessa Sochat.
# This Source Code Form is subject to the terms of the
# Mozilla Public License, v. 2.0. If a copy of the MPL was not distributed
# with this file, You can obtain one at http://mozilla.org/MPL/2.0/.
import json
def state(
self, container_id=None, sudo=None, sync_socket=None, singularity_options=None
):
"""get the state of an OciImage, if it exists. The optional states that
can be returned are created, running, stopped or (not existing).
Equivalent command line example:
singularity oci state <container_ID>
Parameters
==========
container_id: the id to get the state of.
sudo: Add sudo to the command. If the container was created by root,
you need sudo to interact and get its state.
sync_socket: the path to the unix socket for state synchronization
singularity_options: a list of options to provide to the singularity client
Returns
=======
state: a parsed json of the container state, if exists. If the
container is not found, None is returned.
"""
sudo = self._get_sudo(sudo)
container_id = self.get_container_id(container_id)
# singularity oci state
cmd = self._init_command("state", singularity_options)
if sync_socket is not None:
cmd = cmd + ["--sync-socket", sync_socket]
# Finally, add the container_id
cmd.append(container_id)
# Get the instance state
result = self._run_command(cmd, sudo=sudo, quiet=True)
if result is not None:
# If successful, a string is returned to parse
if isinstance(result, str):
return json.loads(result)
def _state_command(
self, container_id=None, command="start", sudo=None, singularity_options=None
):
"""A generic state command to wrap pause, resume, kill, etc., where the
only difference is the command. This function will be unwrapped if the
child functions get more complicated (with additional arguments).
Equivalent command line example:
singularity oci <command> <container_ID>
Parameters
==========
container_id: the id to start.
command: one of start, resume, pause, kill, defaults to start.
singularity_options: a list of options to provide to the singularity client
sudo: Add sudo to the command. If the container was created by root,
you need sudo to interact and get its state.
Returns
=======
return_code: the return code to indicate if the container was started.
"""
sudo = self._get_sudo(sudo)
container_id = self.get_container_id(container_id)
# singularity oci state
cmd = self._init_command(command, singularity_options)
# Finally, add the container_id
cmd.append(container_id)
# Run the command, return return code
return self._run_and_return(cmd, sudo)
def start(self, container_id=None, sudo=None, singularity_options=None):
"""start a previously invoked OciImage, if it exists.
Equivalent command line example:
singularity oci start <container_ID>
Parameters
==========
container_id: the id to start.
sudo: Add sudo to the command. If the container was created by root,
you need sudo to interact and get its state.
Returns
=======
return_code: the return code to indicate if the container was started.
"""
return self._state_command(
container_id, sudo=sudo, singularity_options=singularity_options
)
def kill(self, container_id=None, sudo=None, signal=None, singularity_options=None):
"""stop (kill) a started OciImage container, if it exists
Equivalent command line example:
singularity oci kill <container_ID>
Parameters
==========
container_id: the id to stop.
signal: signal sent to the container (default SIGTERM)
singularity_options: a list of options to provide to the singularity client
sudo: Add sudo to the command. If the container was created by root,
you need sudo to interact and get its state.
Returns
=======
return_code: the return code to indicate if the container was killed.
"""
sudo = self._get_sudo(sudo)
container_id = self.get_container_id(container_id)
# singularity oci state
cmd = self._init_command("kill", singularity_options)
# Finally, add the container_id
cmd.append(container_id)
# Add the signal, if defined
if signal is not None:
cmd = cmd + ["--signal", signal]
# Run the command, return return code
return self._run_and_return(cmd, sudo)
def resume(self, container_id=None, sudo=None, singularity_options=None):
"""resume a stopped OciImage container, if it exists
Equivalent command line example:
singularity oci resume <container_ID>
Parameters
==========
container_id: the id to stop.
singularity_options: a list of options to provide to the singularity client
sudo: Add sudo to the command. If the container was created by root,
you need sudo to interact and get its state.
Returns
=======
return_code: the return code to indicate if the container was resumed.
"""
return self._state_command(
container_id,
command="resume",
sudo=sudo,
singularity_options=singularity_options,
)
def pause(self, container_id=None, sudo=None, singularity_options=None):
"""pause a running OciImage container, if it exists
Equivalent command line example:
singularity oci pause <container_ID>
Parameters
==========
container_id: the id to stop.
singularity_options: a list of options to provide to the singularity client
sudo: Add sudo to the command. If the container was created by root,
you need sudo to interact and get its state.
Returns
=======
return_code: the return code to indicate if the container was paused.
"""
return self._state_command(
container_id,
command="pause",
sudo=sudo,
singularity_options=singularity_options,
)
|
