This repository has been archived by the owner on Mar 9, 2022. It is now read-only.
-
Notifications
You must be signed in to change notification settings - Fork 348
/
status.go
247 lines (225 loc) · 7.52 KB
/
status.go
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
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
/*
Copyright The containerd Authors.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/
package container
import (
"encoding/json"
"io/ioutil"
"os"
"path/filepath"
"sync"
"github.com/containerd/continuity"
"github.com/pkg/errors"
runtime "k8s.io/cri-api/pkg/apis/runtime/v1alpha2"
)
// The container state machine in the CRI plugin:
//
// + +
// | |
// | Create | Load
// | |
// +----v----+ |
// | | |
// | CREATED <---------+-----------+
// | | | |
// +----+----- | |
// | | |
// | Start | |
// | | |
// +----v----+ | |
// Exec +--------+ | | |
// Attach | | RUNNING <---------+ |
// LogReopen +--------> | | |
// +----+----+ | |
// | | |
// | Stop/Exit | |
// | | |
// +----v----+ | |
// | <---------+ +----v----+
// | EXITED | | |
// | <----------------+ UNKNOWN |
// +----+----+ Stop | |
// | +---------+
// | Remove
// v
// DELETED
// statusVersion is current version of container status.
const statusVersion = "v1" // nolint
// versionedStatus is the internal used versioned container status.
// nolint
type versionedStatus struct {
// Version indicates the version of the versioned container status.
Version string
Status
}
// Status is the status of a container.
type Status struct {
// Pid is the init process id of the container.
Pid uint32
// CreatedAt is the created timestamp.
CreatedAt int64
// StartedAt is the started timestamp.
StartedAt int64
// FinishedAt is the finished timestamp.
FinishedAt int64
// ExitCode is the container exit code.
ExitCode int32
// CamelCase string explaining why container is in its current state.
Reason string
// Human-readable message indicating details about why container is in its
// current state.
Message string
// Starting indicates that the container is in starting state.
// This field doesn't need to be checkpointed.
Starting bool `json:"-"`
// Removing indicates that the container is in removing state.
// This field doesn't need to be checkpointed.
Removing bool `json:"-"`
// Unknown indicates that the container status is not fully loaded.
// This field doesn't need to be checkpointed.
Unknown bool `json:"-"`
}
// State returns current state of the container based on the container status.
func (s Status) State() runtime.ContainerState {
if s.Unknown {
return runtime.ContainerState_CONTAINER_UNKNOWN
}
if s.FinishedAt != 0 {
return runtime.ContainerState_CONTAINER_EXITED
}
if s.StartedAt != 0 {
return runtime.ContainerState_CONTAINER_RUNNING
}
if s.CreatedAt != 0 {
return runtime.ContainerState_CONTAINER_CREATED
}
return runtime.ContainerState_CONTAINER_UNKNOWN
}
// encode encodes Status into bytes in json format.
func (s *Status) encode() ([]byte, error) {
return json.Marshal(&versionedStatus{
Version: statusVersion,
Status: *s,
})
}
// decode decodes Status from bytes.
func (s *Status) decode(data []byte) error {
versioned := &versionedStatus{}
if err := json.Unmarshal(data, versioned); err != nil {
return err
}
// Handle old version after upgrade.
switch versioned.Version {
case statusVersion:
*s = versioned.Status
return nil
}
return errors.New("unsupported version")
}
// UpdateFunc is function used to update the container status. If there
// is an error, the update will be rolled back.
type UpdateFunc func(Status) (Status, error)
// StatusStorage manages the container status with a storage backend.
type StatusStorage interface {
// Get a container status.
Get() Status
// UpdateSync updates the container status and the on disk checkpoint.
// Note that the update MUST be applied in one transaction.
UpdateSync(UpdateFunc) error
// Update the container status. Note that the update MUST be applied
// in one transaction.
Update(UpdateFunc) error
// Delete the container status.
// Note:
// * Delete should be idempotent.
// * The status must be deleted in one trasaction.
Delete() error
}
// StoreStatus creates the storage containing the passed in container status with the
// specified id.
// The status MUST be created in one transaction.
func StoreStatus(root, id string, status Status) (StatusStorage, error) {
data, err := status.encode()
if err != nil {
return nil, errors.Wrap(err, "failed to encode status")
}
path := filepath.Join(root, "status")
if err := continuity.AtomicWriteFile(path, data, 0600); err != nil {
return nil, errors.Wrapf(err, "failed to checkpoint status to %q", path)
}
return &statusStorage{
path: path,
status: status,
}, nil
}
// LoadStatus loads container status from checkpoint. There shouldn't be threads
// writing to the file during loading.
func LoadStatus(root, id string) (Status, error) {
path := filepath.Join(root, "status")
data, err := ioutil.ReadFile(path)
if err != nil {
return Status{}, errors.Wrapf(err, "failed to read status from %q", path)
}
var status Status
if err := status.decode(data); err != nil {
return Status{}, errors.Wrapf(err, "failed to decode status %q", data)
}
return status, nil
}
type statusStorage struct {
sync.RWMutex
path string
status Status
}
// Get a copy of container status.
func (s *statusStorage) Get() Status {
s.RLock()
defer s.RUnlock()
return s.status
}
// UpdateSync updates the container status and the on disk checkpoint.
func (s *statusStorage) UpdateSync(u UpdateFunc) error {
s.Lock()
defer s.Unlock()
newStatus, err := u(s.status)
if err != nil {
return err
}
data, err := newStatus.encode()
if err != nil {
return errors.Wrap(err, "failed to encode status")
}
if err := continuity.AtomicWriteFile(s.path, data, 0600); err != nil {
return errors.Wrapf(err, "failed to checkpoint status to %q", s.path)
}
s.status = newStatus
return nil
}
// Update the container status.
func (s *statusStorage) Update(u UpdateFunc) error {
s.Lock()
defer s.Unlock()
newStatus, err := u(s.status)
if err != nil {
return err
}
s.status = newStatus
return nil
}
// Delete deletes the container status from disk atomically.
func (s *statusStorage) Delete() error {
temp := filepath.Dir(s.path) + ".del-" + filepath.Base(s.path)
if err := os.Rename(s.path, temp); err != nil && !os.IsNotExist(err) {
return err
}
return os.RemoveAll(temp)
}