github/containerd
2
0
Fork 0
Code Issues Pull Requests Packages Projects Releases 2 Wiki Activity

Remove old metadata store.

Browse Source 
Signed-off-by: Lantao Liu <lantaol@google.com>
This commit is contained in:
Lantao Liu 2017-06-06 18:02:09 +00:00
parent 7b16a35287
commit f4df66eaaf
10 changed files with 0 additions and 1406 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

183
pkg/metadata/container.go
View File

@ -1,183 +0,0 @@
/*
Copyright 2017 The Kubernetes 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 metadata
import (
"encoding/json"
"k8s.io/kubernetes/pkg/kubelet/apis/cri/v1alpha1/runtime"
"github.com/kubernetes-incubator/cri-containerd/pkg/metadata/store"
)
// The code is very similar with sandbox.go, but there is no template support
// in golang, we have to have similar files for different types.
// TODO(random-liu): Figure out a way to simplify this.
// TODO(random-liu): Handle versioning with the same mechanism with container.go
// containerMetadataVersion is current version of container metadata.
const containerMetadataVersion = "v1" // nolint
// versionedContainerMetadata is the internal versioned container metadata.
// nolint
type versionedContainerMetadata struct {
// Version indicates the version of the versioned container metadata.
Version string
ContainerMetadata
}
// ContainerMetadata is the unversioned container metadata.
type ContainerMetadata struct {
// ID is the container id.
ID string
// Name is the container name.
Name string
// SandboxID is the sandbox id the container belongs to.
SandboxID string
// Config is the CRI container config.
Config *runtime.ContainerConfig
// ImageRef is the reference of image used by the container.
ImageRef string
// 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
// Removing indicates that the container is in removing state.
// In fact, this field doesn't need to be checkpointed.
// TODO(random-liu): Skip this during serialization when we put object
// into the store directly.
// TODO(random-liu): Reset this field to false during state recovery.
Removing bool
}
// State returns current state of the container based on the metadata.
func (c *ContainerMetadata) State() runtime.ContainerState {
if c.FinishedAt != 0 {
return runtime.ContainerState_CONTAINER_EXITED
}
if c.StartedAt != 0 {
return runtime.ContainerState_CONTAINER_RUNNING
}
if c.CreatedAt != 0 {
return runtime.ContainerState_CONTAINER_CREATED
}
return runtime.ContainerState_CONTAINER_UNKNOWN
}
// ContainerUpdateFunc is the function used to update ContainerMetadata.
type ContainerUpdateFunc func(ContainerMetadata) (ContainerMetadata, error)
// ContainerToStoreUpdateFunc generates a metadata store UpdateFunc from ContainerUpdateFunc.
func ContainerToStoreUpdateFunc(u ContainerUpdateFunc) store.UpdateFunc {
return func(data []byte) ([]byte, error) {
meta := &ContainerMetadata{}
if err := json.Unmarshal(data, meta); err != nil {
return nil, err
}
newMeta, err := u(*meta)
if err != nil {
return nil, err
}
return json.Marshal(newMeta)
}
}
// ContainerStore is the store for metadata of all containers.
type ContainerStore interface {
// Create creates a container from ContainerMetadata in the store.
Create(ContainerMetadata) error
// Get gets a specified container.
Get(string) (*ContainerMetadata, error)
// Update updates a specified container.
Update(string, ContainerUpdateFunc) error
// List lists all containers.
List() ([]*ContainerMetadata, error)
// Delete deletes the container from the store.
Delete(string) error
}
// containerStore is an implmentation of ContainerStore.
type containerStore struct {
store store.MetadataStore
}
// NewContainerStore creates a ContainerStore from a basic MetadataStore.
func NewContainerStore(store store.MetadataStore) ContainerStore {
return &containerStore{store: store}
}
// Create creates a container from ContainerMetadata in the store.
func (c *containerStore) Create(metadata ContainerMetadata) error {
data, err := json.Marshal(&metadata)
if err != nil {
return err
}
return c.store.Create(metadata.ID, data)
}
// Get gets a specified container.
func (c *containerStore) Get(containerID string) (*ContainerMetadata, error) {
data, err := c.store.Get(containerID)
if err != nil {
return nil, err
}
container := &ContainerMetadata{}
if err := json.Unmarshal(data, container); err != nil {
return nil, err
}
return container, nil
}
// Update updates a specified container. The function is running in a
// transaction. Update will not be applied when the update function
// returns error.
func (c *containerStore) Update(containerID string, u ContainerUpdateFunc) error {
return c.store.Update(containerID, ContainerToStoreUpdateFunc(u))
}
// List lists all containers.
func (c *containerStore) List() ([]*ContainerMetadata, error) {
allData, err := c.store.List()
if err != nil {
return nil, err
}
var containers []*ContainerMetadata
for _, data := range allData {
container := &ContainerMetadata{}
if err := json.Unmarshal(data, container); err != nil {
return nil, err
}
containers = append(containers, container)
}
return containers, nil
}
// Delete deletes the Container from the store.
func (c *containerStore) Delete(containerID string) error {
return c.store.Delete(containerID)
}

180
pkg/metadata/container_test.go
View File

@ -1,180 +0,0 @@
/*
Copyright 2017 The Kubernetes Authorc.
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 metadata
import (
"testing"
"time"
assertlib "github.com/stretchr/testify/assert"
"github.com/kubernetes-incubator/cri-containerd/pkg/metadata/store"
"k8s.io/kubernetes/pkg/kubelet/apis/cri/v1alpha1/runtime"
)
func TestContainerState(t *testing.T) {
for c, test := range map[string]struct {
metadata *ContainerMetadata
state runtime.ContainerState
}{
"unknown state": {
metadata: &ContainerMetadata{
ID: "1",
Name: "Container-1",
},
state: runtime.ContainerState_CONTAINER_UNKNOWN,
},
"created state": {
metadata: &ContainerMetadata{
ID: "2",
Name: "Container-2",
CreatedAt: time.Now().UnixNano(),
},
state: runtime.ContainerState_CONTAINER_CREATED,
},
"running state": {
metadata: &ContainerMetadata{
ID: "3",
Name: "Container-3",
CreatedAt: time.Now().UnixNano(),
StartedAt: time.Now().UnixNano(),
},
state: runtime.ContainerState_CONTAINER_RUNNING,
},
"exited state": {
metadata: &ContainerMetadata{
ID: "3",
Name: "Container-3",
CreatedAt: time.Now().UnixNano(),
FinishedAt: time.Now().UnixNano(),
},
state: runtime.ContainerState_CONTAINER_EXITED,
},
} {
t.Logf("TestCase %q", c)
assertlib.Equal(t, test.state, test.metadata.State())
}
}
func TestContainerStore(t *testing.T) {
containers := map[string]*ContainerMetadata{
"1": {
ID: "1",
Name: "Container-1",
SandboxID: "Sandbox-1",
Config: &runtime.ContainerConfig{
Metadata: &runtime.ContainerMetadata{
Name: "TestPod-1",
Attempt: 1,
},
},
ImageRef: "TestImage-1",
Pid: 1,
CreatedAt: time.Now().UnixNano(),
StartedAt: time.Now().UnixNano(),
FinishedAt: time.Now().UnixNano(),
ExitCode: 1,
Reason: "TestReason-1",
Message: "TestMessage-1",
},
"2": {
ID: "2",
Name: "Container-2",
SandboxID: "Sandbox-2",
Config: &runtime.ContainerConfig{
Metadata: &runtime.ContainerMetadata{
Name: "TestPod-2",
Attempt: 2,
},
},
ImageRef: "TestImage-2",
Pid: 2,
CreatedAt: time.Now().UnixNano(),
StartedAt: time.Now().UnixNano(),
FinishedAt: time.Now().UnixNano(),
ExitCode: 2,
Reason: "TestReason-2",
Message: "TestMessage-2",
},
"3": {
ID: "3",
Name: "Container-3",
SandboxID: "Sandbox-3",
Config: &runtime.ContainerConfig{
Metadata: &runtime.ContainerMetadata{
Name: "TestPod-3",
Attempt: 3,
},
},
ImageRef: "TestImage-3",
Pid: 3,
CreatedAt: time.Now().UnixNano(),
StartedAt: time.Now().UnixNano(),
FinishedAt: time.Now().UnixNano(),
ExitCode: 3,
Reason: "TestReason-3",
Message: "TestMessage-3",
Removing: true,
},
}
assert := assertlib.New(t)
c := NewContainerStore(store.NewMetadataStore())
t.Logf("should be able to create container metadata")
for _, meta := range containers {
assert.NoError(c.Create(*meta))
}
t.Logf("should be able to get container metadata")
for id, expectMeta := range containers {
meta, err := c.Get(id)
assert.NoError(err)
assert.Equal(expectMeta, meta)
}
t.Logf("should be able to list container metadata")
cntrs, err := c.List()
assert.NoError(err)
assert.Len(cntrs, 3)
t.Logf("should be able to update container metadata")
testID := "2"
newCreatedAt := time.Now().UnixNano()
expectMeta := *containers[testID]
expectMeta.CreatedAt = newCreatedAt
err = c.Update(testID, func(o ContainerMetadata) (ContainerMetadata, error) {
o.CreatedAt = newCreatedAt
return o, nil
})
assert.NoError(err)
newMeta, err := c.Get(testID)
assert.NoError(err)
assert.Equal(&expectMeta, newMeta)
t.Logf("should be able to delete container metadata")
assert.NoError(c.Delete(testID))
cntrs, err = c.List()
assert.NoError(err)
assert.Len(cntrs, 2)
t.Logf("get should return nil without error after deletion")
meta, err := c.Get(testID)
assert.Error(store.ErrNotExist, err)
assert.True(meta == nil)
}

151
pkg/metadata/image_metadata.go
View File

@ -1,151 +0,0 @@
/*
Copyright 2017 The Kubernetes 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 metadata
import (
"encoding/json"
imagespec "github.com/opencontainers/image-spec/specs-go/v1"
"github.com/kubernetes-incubator/cri-containerd/pkg/metadata/store"
)
// The code is very similar to sandbox.go, but there is no template support
// in golang, thus similar files for different types.
// TODO(random-liu): Figure out a way to simplify this.
// TODO(random-liu): Handle versioning
// imageMetadataVersion is current version of image metadata.
const imageMetadataVersion = "v1" // nolint
// versionedImageMetadata is the internal struct representing the versioned
// image metadata
// nolint
type versionedImageMetadata struct {
// Version indicates the version of the versioned image metadata.
Version string `json:"version,omitempty"`
ImageMetadata
}
// ImageMetadata is the unversioned image metadata.
type ImageMetadata struct {
// Id of the image. Normally the digest of image config.
ID string `json:"id,omitempty"`
// ChainID is the chainID of the image.
ChainID string `json:"chain_id,omitempty"`
// Other names by which this image is known.
RepoTags []string `json:"repo_tags,omitempty"`
// Digests by which this image is known.
RepoDigests []string `json:"repo_digests,omitempty"`
// Size is the compressed size of the image.
Size int64 `json:"size,omitempty"`
// Config is the oci image config of the image.
Config *imagespec.ImageConfig `json:"config,omitempty"`
}
// ImageMetadataUpdateFunc is the function used to update ImageMetadata.
type ImageMetadataUpdateFunc func(ImageMetadata) (ImageMetadata, error)
// imageMetadataToStoreUpdateFunc generates a metadata store UpdateFunc from ImageMetadataUpdateFunc.
func imageMetadataToStoreUpdateFunc(u ImageMetadataUpdateFunc) store.UpdateFunc {
return func(data []byte) ([]byte, error) {
meta := &ImageMetadata{}
if err := json.Unmarshal(data, meta); err != nil {
return nil, err
}
newMeta, err := u(*meta)
if err != nil {
return nil, err
}
return json.Marshal(newMeta)
}
}
// ImageMetadataStore is the store for metadata of all images.
type ImageMetadataStore interface {
// Create creates an image's metadata from ImageMetadata in the store.
Create(ImageMetadata) error
// Get gets the specified image metadata.
Get(string) (*ImageMetadata, error)
// Update updates a specified image metatdata.
Update(string, ImageMetadataUpdateFunc) error
// List lists all image metadatas.
List() ([]*ImageMetadata, error)
// Delete deletes the image's metatdata from the store.
Delete(string) error
}
// imageMetadataStore is an implmentation of ImageMetadataStore.
type imageMetadataStore struct {
store store.MetadataStore
}
// NewImageMetadataStore creates an ImageMetadataStore from a basic MetadataStore.
func NewImageMetadataStore(store store.MetadataStore) ImageMetadataStore {
return &imageMetadataStore{store: store}
}
// Create creates a image's metadata from ImageMetadata in the store.
func (s *imageMetadataStore) Create(metadata ImageMetadata) error {
data, err := json.Marshal(&metadata)
if err != nil {
return err
}
return s.store.Create(metadata.ID, data)
}
// Get gets the specified image metadata.
func (s *imageMetadataStore) Get(digest string) (*ImageMetadata, error) {
data, err := s.store.Get(digest)
if err != nil {
return nil, err
}
imageMetadata := &ImageMetadata{}
if err := json.Unmarshal(data, imageMetadata); err != nil {
return nil, err
}
return imageMetadata, nil
}
// Update updates a specified image's metadata. The function is running in a
// transaction. Update will not be applied when the update function
// returns error.
func (s *imageMetadataStore) Update(digest string, u ImageMetadataUpdateFunc) error {
return s.store.Update(digest, imageMetadataToStoreUpdateFunc(u))
}
// List lists all image metadata.
func (s *imageMetadataStore) List() ([]*ImageMetadata, error) {
allData, err := s.store.List()
if err != nil {
return nil, err
}
var imageMetadataA []*ImageMetadata
for _, data := range allData {
imageMetadata := &ImageMetadata{}
if err := json.Unmarshal(data, imageMetadata); err != nil {
return nil, err
}
imageMetadataA = append(imageMetadataA, imageMetadata)
}
return imageMetadataA, nil
}
// Delete deletes the image metadata from the store.
func (s *imageMetadataStore) Delete(digest string) error {
return s.store.Delete(digest)
}

100
pkg/metadata/image_metadata_test.go
View File

@ -1,100 +0,0 @@
/*
Copyright 2017 The Kubernetes 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 metadata
import (
"testing"
imagespec "github.com/opencontainers/image-spec/specs-go/v1"
assertlib "github.com/stretchr/testify/assert"
"github.com/kubernetes-incubator/cri-containerd/pkg/metadata/store"
)
func TestImageMetadataStore(t *testing.T) {
imageMetadataMap := map[string]*ImageMetadata{
"1": {
ID: "1",
ChainID: "test-chain-id-1",
RepoTags: []string{"tag-1"},
RepoDigests: []string{"digest-1"},
Size: 10,
Config: &imagespec.ImageConfig{},
},
"2": {
ID: "2",
ChainID: "test-chain-id-2",
RepoTags: []string{"tag-2"},
RepoDigests: []string{"digest-2"},
Size: 20,
Config: &imagespec.ImageConfig{},
},
"3": {
ID: "3",
RepoTags: []string{"tag-3"},
RepoDigests: []string{"digest-3"},
ChainID: "test-chain-id-3",
Size: 30,
Config: &imagespec.ImageConfig{},
},
}
assert := assertlib.New(t)
s := NewImageMetadataStore(store.NewMetadataStore())
t.Logf("should be able to create image metadata")
for _, meta := range imageMetadataMap {
assert.NoError(s.Create(*meta))
}
t.Logf("should be able to get image metadata")
for id, expectMeta := range imageMetadataMap {
meta, err := s.Get(id)
assert.NoError(err)
assert.Equal(expectMeta, meta)
}
t.Logf("should be able to list image metadata")
imgs, err := s.List()
assert.NoError(err)
assert.Len(imgs, 3)
t.Logf("should be able to update image metadata")
testID := "2"
newSize := int64(200)
expectMeta := *imageMetadataMap[testID]
expectMeta.Size = newSize
err = s.Update(testID, func(o ImageMetadata) (ImageMetadata, error) {
o.Size = newSize
return o, nil
})
assert.NoError(err)
newMeta, err := s.Get(testID)
assert.NoError(err)
assert.Equal(&expectMeta, newMeta)
t.Logf("should be able to delete image metadata")
assert.NoError(s.Delete(testID))
imgs, err = s.List()
assert.NoError(err)
assert.Len(imgs, 2)
t.Logf("get should return nil not exist error after deletion")
meta, err := s.Get(testID)
assert.Error(store.ErrNotExist, err)
assert.Nil(meta)
}

154
pkg/metadata/sandbox.go
View File

@ -1,154 +0,0 @@
/*
Copyright 2017 The Kubernetes 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 metadata
import (
"encoding/json"
"github.com/kubernetes-incubator/cri-containerd/pkg/metadata/store"
"k8s.io/kubernetes/pkg/kubelet/apis/cri/v1alpha1/runtime"
)
// TODO(random-liu): Handle versioning around all marshal/unmarshal.
// version and versionedSandboxMetadata is not used now, but should be used
// in the future:
// 1) Only versioned metadata should be written into the store;
// 2) Only unversioned metadata should be returned to the user;
// 3) A conversion function is needed to convert any supported versioned
// metadata into unversioned metadata.
// sandboxMetadataVersion is current version of sandbox metadata.
const sandboxMetadataVersion = "v1" // nolint
// versionedSandboxMetadata is the internal struct representing the versioned
// sandbox metadata
// nolint
type versionedSandboxMetadata struct {
// Version indicates the version of the versioned sandbox metadata.
Version string
SandboxMetadata
}
// SandboxMetadata is the unversioned sandbox metadata.
type SandboxMetadata struct {
// ID is the sandbox id.
ID string
// Name is the sandbox name.
Name string
// Config is the CRI sandbox config.
Config *runtime.PodSandboxConfig
// CreatedAt is the created timestamp.
CreatedAt int64
// NetNS is the network namespace used by the sandbox.
NetNS string
// Pid is the process id of the sandbox.
Pid uint32
}
// SandboxUpdateFunc is the function used to update SandboxMetadata.
type SandboxUpdateFunc func(SandboxMetadata) (SandboxMetadata, error)
// sandboxToStoreUpdateFunc generates a metadata store UpdateFunc from SandboxUpdateFunc.
func sandboxToStoreUpdateFunc(u SandboxUpdateFunc) store.UpdateFunc {
return func(data []byte) ([]byte, error) {
meta := &SandboxMetadata{}
if err := json.Unmarshal(data, meta); err != nil {
return nil, err
}
newMeta, err := u(*meta)
if err != nil {
return nil, err
}
return json.Marshal(newMeta)
}
}
// SandboxStore is the store for metadata of all sandboxes.
type SandboxStore interface {
// Create creates a sandbox from SandboxMetadata in the store.
Create(SandboxMetadata) error
// Get gets the specified sandbox.
Get(string) (*SandboxMetadata, error)
// Update updates a specified sandbox.
Update(string, SandboxUpdateFunc) error
// List lists all sandboxes.
List() ([]*SandboxMetadata, error)
// Delete deletes the sandbox from the store.
Delete(string) error
}
// sandboxStore is an implmentation of SandboxStore.
type sandboxStore struct {
store store.MetadataStore
}
// NewSandboxStore creates a SandboxStore from a basic MetadataStore.
func NewSandboxStore(store store.MetadataStore) SandboxStore {
return &sandboxStore{store: store}
}
// Create creates a sandbox from SandboxMetadata in the store.
func (s *sandboxStore) Create(metadata SandboxMetadata) error {
data, err := json.Marshal(&metadata)
if err != nil {
return err
}
return s.store.Create(metadata.ID, data)
}
// Get gets the specified sandbox.
func (s *sandboxStore) Get(sandboxID string) (*SandboxMetadata, error) {
data, err := s.store.Get(sandboxID)
if err != nil {
return nil, err
}
sandbox := &SandboxMetadata{}
if err := json.Unmarshal(data, sandbox); err != nil {
return nil, err
}
return sandbox, nil
}
// Update updates a specified sandbox. The function is running in a
// transaction. Update will not be applied when the update function
// returns error.
func (s *sandboxStore) Update(sandboxID string, u SandboxUpdateFunc) error {
return s.store.Update(sandboxID, sandboxToStoreUpdateFunc(u))
}
// List lists all sandboxes.
func (s *sandboxStore) List() ([]*SandboxMetadata, error) {
allData, err := s.store.List()
if err != nil {
return nil, err
}
var sandboxes []*SandboxMetadata
for _, data := range allData {
sandbox := &SandboxMetadata{}
if err := json.Unmarshal(data, sandbox); err != nil {
return nil, err
}
sandboxes = append(sandboxes, sandbox)
}
return sandboxes, nil
}
// Delete deletes the sandbox from the store.
func (s *sandboxStore) Delete(sandboxID string) error {
return s.store.Delete(sandboxID)
}

123
pkg/metadata/sandbox_test.go
View File

@ -1,123 +0,0 @@
/*
Copyright 2017 The Kubernetes 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 metadata
import (
"testing"
"time"
assertlib "github.com/stretchr/testify/assert"
"github.com/kubernetes-incubator/cri-containerd/pkg/metadata/store"
"k8s.io/kubernetes/pkg/kubelet/apis/cri/v1alpha1/runtime"
)
func TestSandboxStore(t *testing.T) {
sandboxes := map[string]*SandboxMetadata{
"1": {
ID: "1",
Name: "Sandbox-1",
Config: &runtime.PodSandboxConfig{
Metadata: &runtime.PodSandboxMetadata{
Name: "TestPod-1",
Uid: "TestUid-1",
Namespace: "TestNamespace-1",
Attempt: 1,
},
},
CreatedAt: time.Now().UnixNano(),
NetNS: "TestNetNS-1",
Pid: 1001,
},
"2": {
ID: "2",
Name: "Sandbox-2",
Config: &runtime.PodSandboxConfig{
Metadata: &runtime.PodSandboxMetadata{
Name: "TestPod-2",
Uid: "TestUid-2",
Namespace: "TestNamespace-2",
Attempt: 2,
},
},
CreatedAt: time.Now().UnixNano(),
NetNS: "TestNetNS-2",
Pid: 1002,
},
"3": {
ID: "3",
Name: "Sandbox-3",
Config: &runtime.PodSandboxConfig{
Metadata: &runtime.PodSandboxMetadata{
Name: "TestPod-3",
Uid: "TestUid-3",
Namespace: "TestNamespace-3",
Attempt: 3,
},
},
CreatedAt: time.Now().UnixNano(),
NetNS: "TestNetNS-3",
Pid: 1003,
},
}
assert := assertlib.New(t)
s := NewSandboxStore(store.NewMetadataStore())
t.Logf("should be able to create sandbox metadata")
for _, meta := range sandboxes {
assert.NoError(s.Create(*meta))
}
t.Logf("should be able to get sandbox metadata")
for id, expectMeta := range sandboxes {
meta, err := s.Get(id)
assert.NoError(err)
assert.Equal(expectMeta, meta)
}
t.Logf("should be able to list sandbox metadata")
sbs, err := s.List()
assert.NoError(err)
assert.Len(sbs, 3)
t.Logf("should be able to update sandbox metadata")
testID := "2"
newCreatedAt := time.Now().UnixNano()
expectMeta := *sandboxes[testID]
expectMeta.CreatedAt = newCreatedAt
err = s.Update(testID, func(o SandboxMetadata) (SandboxMetadata, error) {
o.CreatedAt = newCreatedAt
return o, nil
})
assert.NoError(err)
newMeta, err := s.Get(testID)
assert.NoError(err)
assert.Equal(&expectMeta, newMeta)
t.Logf("should be able to delete sandbox metadata")
assert.NoError(s.Delete(testID))
sbs, err = s.List()
assert.NoError(err)
assert.Len(sbs, 2)
t.Logf("get should return nil with not exist error after deletion")
meta, err := s.Get(testID)
assert.Error(store.ErrNotExist, err)
assert.Nil(meta)
}

246
pkg/metadata/store/metadata_store.go
View File

@ -1,246 +0,0 @@
/*
Copyright 2017 The Kubernetes 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 store
import (
"errors"
"sync"
"github.com/golang/glog"
)
var (
// ErrNotExist is the error returned when specified id does
// not exist.
ErrNotExist = errors.New("does not exist")
// ErrAlreadyExist is the error returned when specified id already
// exists.
ErrAlreadyExist = errors.New("already exists")
)
// All byte arrays are expected to be read-only. User MUST NOT modify byte
// array element directly!!
// UpdateFunc is function used to update a specific metadata. The value
// passed in is the old value, it MUST NOT be changed in the function.
// The function should make a copy of the old value and apply update on
// the copy. The updated value should be returned. If there is an error,
// the update will be rolled back.
type UpdateFunc func([]byte) ([]byte, error)
// MetadataStore is the interface for storing metadata. All methods should
// be thread-safe.
// TODO(random-liu): Initialize the metadata store with a type, and replace
// []byte with interface{}, so as to avoid extra marshal/unmarshal on the
// user side.
type MetadataStore interface {
// Create the metadata containing the passed in data with the
// specified id.
// Note:
// * Create MUST return error if the id already exists.
// * The id and data MUST be added in one transaction to the store.
Create(string, []byte) error
// Get the data by id.
// Note that Get MUST return ErrNotExist if the id doesn't exist.
Get(string) ([]byte, error)
// Update the data by id.
// Note:
// * Update MUST return ErrNotExist is the id doesn't exist.
// * The update MUST be applied in one transaction.
Update(string, UpdateFunc) error
// List returns entire array of data from the store.
List() ([][]byte, error)
// Delete the data by id.
// Note:
// * Delete should be idempotent, it MUST not return error if the id
// doesn't exist or has been removed.
// * The id and data MUST be deleted in one transaction.
Delete(string) error
}
// TODO(random-liu) Add checkpoint. When checkpoint is enabled, it should cache data
// in memory and checkpoint metadata into files during update. Metadata should serve
// from memory, but any modification should be checkpointed, so that memory could be
// recovered after restart. It should be possible to disable the checkpoint for testing.
// Note that checkpoint update may fail, so the recovery logic should tolerate that.
// metadata is the internal type for storing data in metadataStore.
type metadata struct {
sync.RWMutex
data []byte
}
// newMetadata creates a new metadata.
func newMetadata(data []byte) (*metadata, error) {
return &metadata{data: data}, nil
// TODO(random-liu): Create the data on disk atomically.
}
// get a snapshot of the metadata.
func (m *metadata) get() []byte {
m.RLock()
defer m.RUnlock()
return m.data
}
// update the value.
func (m *metadata) update(u UpdateFunc) error {
m.Lock()
defer m.Unlock()
newData, err := u(m.data)
if err != nil {
return err
}
// Replace with newData, user holding the old data will not
// be affected.
// TODO(random-liu) *Update* existing data on disk atomically,
// return error if checkpoint failed.
m.data = newData
return nil
}
// delete deletes the data on disk atomically.
func (m *metadata) delete() error {
// TODO(random-liu): Hold write lock, rename the data on the disk.
return nil
}
// cleanup cleans up all temporary files left-over.
func (m *metadata) cleanup() error {
// TODO(random-liu): Hold write lock, Cleanup temporary files generated
// in atomic file operations. The write lock makes sure there is no on-going
// update, so any temporary files could be removed.
return nil
}
// metadataStore is metadataStore is an implementation of MetadataStore.
type metadataStore struct {
sync.RWMutex
metas map[string]*metadata
}
// NewMetadataStore creates a MetadataStore.
func NewMetadataStore() MetadataStore {
// TODO(random-liu): Recover state from disk checkpoint.
// TODO(random-liu): Cleanup temporary files left over.
return &metadataStore{metas: map[string]*metadata{}}
}
// createMetadata creates metadata with a read-write lock
func (m *metadataStore) createMetadata(id string, meta *metadata) error {
m.Lock()
defer m.Unlock()
if _, found := m.metas[id]; found {
return ErrAlreadyExist
}
m.metas[id] = meta
return nil
}
// Create the metadata with a specific id.
func (m *metadataStore) Create(id string, data []byte) (retErr error) {
// newMetadata takes time, we may not want to lock around it.
meta, err := newMetadata(data)
if err != nil {
return err
}
defer func() {
// This should not happen, because if id already exists,
// newMetadata should fail to checkpoint. Add this just
// in case.
if retErr != nil {
meta.delete() // nolint: errcheck
meta.cleanup() // nolint: errcheck
}
}()
return m.createMetadata(id, meta)
}
// getMetadata gets metadata by id with a read lock.
func (m *metadataStore) getMetadata(id string) (*metadata, bool) {
m.RLock()
defer m.RUnlock()
meta, found := m.metas[id]
return meta, found
}
// Get data by id.
func (m *metadataStore) Get(id string) ([]byte, error) {
meta, found := m.getMetadata(id)
if !found {
return nil, ErrNotExist
}
return meta.get(), nil
}
// Update data by id.
func (m *metadataStore) Update(id string, u UpdateFunc) error {
meta, found := m.getMetadata(id)
if !found {
return ErrNotExist
}
return meta.update(u)
}
// listMetadata lists all metadata with a read lock.
func (m *metadataStore) listMetadata() []*metadata {
m.RLock()
defer m.RUnlock()
var metas []*metadata
for _, meta := range m.metas {
metas = append(metas, meta)
}
return metas
}
// List all data.
func (m *metadataStore) List() ([][]byte, error) {
metas := m.listMetadata()
var data [][]byte
for _, meta := range metas {
data = append(data, meta.get())
}
return data, nil
}
// Delete the data by id.
func (m *metadataStore) Delete(id string) error {
meta, err := func() (*metadata, error) {
m.Lock()
defer m.Unlock()
meta := m.metas[id]
if meta == nil {
return nil, nil
}
if err := meta.delete(); err != nil {
return nil, err
}
delete(m.metas, id)
return meta, nil
}()
if err != nil {
return err
}
// The metadata is removed from the store at this point.
if meta != nil {
// Do not return error for cleanup.
if err := meta.cleanup(); err != nil {
glog.Errorf("Failed to cleanup metadata %q: %v", id, err)
}
}
return nil
}

209
pkg/metadata/store/metadata_store_test.go
View File

@ -1,209 +0,0 @@
/*
Copyright 2017 The Kubernetes 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 store
import (
"bytes"
"errors"
"fmt"
"sync"
"testing"
assertlib "github.com/stretchr/testify/assert"
)
func TestMetadata(t *testing.T) {
testData := [][]byte{
[]byte("test-data-1"),
[]byte("test-data-2"),
}
updateErr := errors.New("update error")
assert := assertlib.New(t)
t.Logf("simple create and get")
meta, err := newMetadata(testData[0])
assert.NoError(err)
old := meta.get()
assert.Equal(testData[0], old)
t.Logf("failed update should not take effect")
err = meta.update(func(in []byte) ([]byte, error) {
return testData[1], updateErr
})
assert.Equal(updateErr, err)
assert.Equal(testData[0], meta.get())
t.Logf("successful update should take effect")
err = meta.update(func(in []byte) ([]byte, error) {
return testData[1], nil
})
assert.NoError(err)
assert.Equal(testData[1], meta.get())
t.Logf("successful update should not affect existing snapshot")
assert.Equal(testData[0], old)
// TODO(random-liu): Test deleteCheckpoint and cleanupCheckpoint after
// disk based implementation is added.
}
func TestMetadataStore(t *testing.T) {
testIds := []string{"id-0", "id-1"}
testMeta := map[string][]byte{
testIds[0]: []byte("metadata-0"),
testIds[1]: []byte("metadata-1"),
}
assert := assertlib.New(t)
m := NewMetadataStore()
t.Logf("should be empty initially")
metas, err := m.List()
assert.NoError(err)
assert.Empty(metas)
t.Logf("should be able to create metadata")
err = m.Create(testIds[0], testMeta[testIds[0]])
assert.NoError(err)
t.Logf("should not be able to create metadata with the same id")
err = m.Create(testIds[0], testMeta[testIds[0]])
assert.Error(err)
t.Logf("should be able to list metadata")
err = m.Create(testIds[1], testMeta[testIds[1]])
assert.NoError(err)
metas, err = m.List()
assert.NoError(err)
assert.True(sliceContainsMap(metas, testMeta))
t.Logf("should be able to get metadata by id")
meta, err := m.Get(testIds[1])
assert.NoError(err)
assert.Equal(testMeta[testIds[1]], meta)
t.Logf("update should take effect")
err = m.Update(testIds[1], func(in []byte) ([]byte, error) {
return []byte("updated-metadata-1"), nil
})
assert.NoError(err)
newMeta, err := m.Get(testIds[1])
assert.NoError(err)
assert.Equal([]byte("updated-metadata-1"), newMeta)
t.Logf("should be able to delete metadata")
assert.NoError(m.Delete(testIds[1]))
metas, err = m.List()
assert.NoError(err)
assert.Len(metas, 1)
assert.Equal(testMeta[testIds[0]], metas[0])
meta, err = m.Get(testIds[1])
assert.Equal(ErrNotExist, err)
assert.Nil(meta)
t.Logf("update should return not exist error after metadata got deleted")
err = m.Update(testIds[1], func(in []byte) ([]byte, error) {
return in, nil
})
assert.Equal(ErrNotExist, err)
t.Logf("existing reference should not be affected by delete")
assert.Equal([]byte("updated-metadata-1"), newMeta)
t.Logf("should be able to reuse the same id after deletion")
err = m.Create(testIds[1], testMeta[testIds[1]])
assert.NoError(err)
}
// sliceMatchMap checks the same elements with a map.
func sliceContainsMap(s [][]byte, m map[string][]byte) bool {
if len(m) != len(s) {
return false
}
for _, expect := range m {
found := false
for _, got := range s {
if bytes.Equal(expect, got) {
found = true
break
}
}
if !found {
return false
}
}
return true
}
func TestMultithreadAccess(t *testing.T) {
m := NewMetadataStore()
assert := assertlib.New(t)
routineNum := 10
var wg sync.WaitGroup
for i := 0; i < routineNum; i++ {
wg.Add(1)
go func(i int) {
id := fmt.Sprintf("%d", i)
t.Logf("should be able to create id %q", id)
expect := []byte(id)
err := m.Create(id, expect)
assert.NoError(err)
got, err := m.Get(id)
assert.NoError(err)
assert.Equal(expect, got)
gotList, err := m.List()
assert.NoError(err)
assert.Contains(gotList, expect)
t.Logf("should be able to update id %q", id)
expect = []byte("update-" + id)
err = m.Update(id, func([]byte) ([]byte, error) {
return expect, nil
})
assert.NoError(err)
got, err = m.Get(id)
assert.NoError(err)
assert.Equal(expect, got)
t.Logf("should be able to delete id %q", id)
err = m.Delete(id)
assert.NoError(err)
got, err = m.Get(id)
assert.Equal(ErrNotExist, err)
assert.Nil(got)
err = m.Update(id, func(in []byte) ([]byte, error) {
return in, nil
})
assert.Equal(ErrNotExist, err)
gotList, err = m.List()
assert.NoError(err)
assert.NotContains(gotList, expect)
wg.Done()
}(i)
}
wg.Wait()
}
// TODO(random-liu): Test recover logic once checkpoint recovery is added.

25
pkg/metadata/util.go
View File

@ -1,25 +0,0 @@
/*
Copyright 2017 The Kubernetes 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 metadata
import "github.com/kubernetes-incubator/cri-containerd/pkg/metadata/store"
// IsNotExistError is a helper function to check whether the error returned
// by metadata store is not exist error.
func IsNotExistError(err error) bool {
return err.Error() == store.ErrNotExist.Error()
}

35
pkg/metadata/util_test.go
View File

@ -1,35 +0,0 @@
/*
Copyright 2017 The Kubernetes 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 metadata
import (
"errors"
"testing"
"github.com/stretchr/testify/assert"
"github.com/kubernetes-incubator/cri-containerd/pkg/metadata/store"
)
func TestIsNotExistError(t *testing.T) {
err := store.ErrNotExist
assert.True(t, IsNotExistError(err))
err = errors.New(store.ErrNotExist.Error())
assert.True(t, IsNotExistError(err))
err = errors.New("random error")
assert.False(t, IsNotExistError(err))
}