Skip to content
Snippets Groups Projects
Unverified Commit ecc8ad8c authored by Kamil Trzcinski's avatar Kamil Trzcinski
Browse files

Better organise code

parent 9437ada7
No related branches found
No related tags found
No related merge requests found
Hide whitespace changes
Inline Side-by-side
blob.go 0 → 100644
+ 14
− 0
View file @ ecc8ad8c
package main
import "path/filepath"
type blobData struct {
name digest
size int64
references int64
etag string
}
func (b *blobData) path() string {
return filepath.Join("blobs", b.name.scopedPath(), "data")
}
blobs.go
+ 2
− 13
View file @ ecc8ad8c
@@ -13,17 +13,6 @@ import (
 
var parallelBlobWalk = flag.Bool("parallel-blob-walk", true, "Allow to use parallel blob walker")
 
type blobData struct {
name digest
size int64
references int64
etag string
}
func (b *blobData) path() string {
return filepath.Join("blobs", b.name.scopedPath(), "data")
}
type blobsData map[digest]*blobData
 
var blobsLock sync.Mutex
@@ -65,7 +54,7 @@ func (b blobsData) sweep() error {
 
for _, blob_ := range b {
blob := blob_
jg.Dispatch(func() error {
jg.dispatch(func() error {
if blob.references > 0 {
return nil
}
@@ -78,7 +67,7 @@ func (b blobsData) sweep() error {
})
}
 
return jg.Finish()
return jg.finish()
}
 
func (b blobsData) addBlob(segments []string, info fileInfo) error {
job_group.go 0 → 100644
+ 34
− 0
View file @ ecc8ad8c
package main
import "sync"
type jobGroup struct {
ch jobsData
wg sync.WaitGroup
err error
lock sync.Mutex
}
func (g *jobGroup) dispatch(fn func() error) {
g.wg.Add(1)
g.ch <- func() {
var err error
defer func() {
if err != nil {
g.lock.Lock()
g.err = err
g.lock.Unlock()
}
}()
defer g.wg.Done()
err = fn()
}
}
func (g *jobGroup) finish() error {
g.wg.Wait()
return g.err
}
jobs.go
+ 2
− 38
View file @ ecc8ad8c
package main
 
import "sync"
type jobsData chan func()
 
var jobsRunner jobsData = make(jobsData)
var parallelWalkRunner jobsData = make(jobsData)
func (ch jobsData) group() *jobsGroup {
return &jobsGroup{ch: ch}
func (ch jobsData) group() *jobGroup {
return &jobGroup{ch: ch}
}
 
func (ch jobsData) run(max int) {
@@ -25,34 +20,3 @@ func (ch jobsData) run(max int) {
func (ch jobsData) close() {
close(ch)
}
type jobsGroup struct {
ch jobsData
wg sync.WaitGroup
err error
lock sync.Mutex
}
func (g *jobsGroup) Dispatch(fn func() error) {
g.wg.Add(1)
g.ch <- func() {
var err error
defer func() {
if err != nil {
g.lock.Lock()
g.err = err
g.lock.Unlock()
}
}()
defer g.wg.Done()
err = fn()
}
}
func (g *jobsGroup) Finish() error {
g.wg.Wait()
return g.err
}
main.go
+ 5
− 0
View file @ ecc8ad8c
@@ -19,6 +19,11 @@ var (
softErrors = flag.Bool("soft-errors", false, "Print errors, but do not fail")
)
 
var (
jobsRunner = make(jobsData)
parallelWalkRunner = make(jobsData)
)
func logErrorln(args ...interface{}) {
if *softErrors {
logrus.Errorln(args...)
manifest.go 0 → 100644
+ 95
− 0
View file @ ecc8ad8c
package main
import (
"encoding/json"
"fmt"
"path/filepath"
"sync"
"github.com/Sirupsen/logrus"
"github.com/docker/distribution"
"github.com/docker/distribution/manifest"
"github.com/docker/distribution/manifest/manifestlist"
"github.com/docker/distribution/manifest/schema1"
"github.com/docker/distribution/manifest/schema2"
)
type manifestData struct {
digest digest
layers []digest
loaded bool
loadErr error
loadLock sync.Mutex
}
func deserializeManifest(data []byte) (distribution.Manifest, error) {
var versioned manifest.Versioned
if err := json.Unmarshal(data, &versioned); err != nil {
return nil, err
}
switch versioned.SchemaVersion {
case 1:
var sm schema1.SignedManifest
err := json.Unmarshal(data, &sm)
return sm, err
case 2:
// This can be an image manifest or a manifest list
switch versioned.MediaType {
case schema2.MediaTypeManifest:
var m schema2.DeserializedManifest
err := json.Unmarshal(data, &m)
return m, err
case manifestlist.MediaTypeManifestList:
var m manifestlist.DeserializedManifestList
err := json.Unmarshal(data, &m)
return m, err
default:
return nil, distribution.ErrManifestVerification{fmt.Errorf("unrecognized manifest content type %s", versioned.MediaType)}
}
}
return nil, fmt.Errorf("unrecognized manifest schema version %d", versioned.SchemaVersion)
}
func (m *manifestData) path() string {
return filepath.Join("blobs", m.digest.scopedPath(), "data")
}
func (m *manifestData) load(blobs blobsData) error {
logrus.Println("MANIFEST:", m.path(), ": loading...")
data, err := currentStorage.Read(m.path(), blobs.etag(m.digest))
if err != nil {
return err
}
manifest, err := deserializeManifest(data)
if err != nil {
return err
}
for _, reference := range manifest.References() {
digest, err := newDigestFromReference([]byte(reference.Digest))
if err != nil {
return err
}
m.layers = append(m.layers, digest)
}
return nil
}
func (m *manifestData) ensureLoaded(blobs blobsData) error {
if !m.loaded {
m.loadLock.Lock()
defer m.loadLock.Unlock()
if !m.loaded {
m.loadErr = m.load(blobs)
m.loaded = true
}
}
return m.loadErr
}
manifests.go
+ 1
− 86
View file @ ecc8ad8c
package main
 
import (
"encoding/json"
"fmt"
"path/filepath"
"sync"
"github.com/Sirupsen/logrus"
"github.com/docker/distribution"
"github.com/docker/distribution/manifest"
"github.com/docker/distribution/manifest/manifestlist"
"github.com/docker/distribution/manifest/schema1"
"github.com/docker/distribution/manifest/schema2"
)
 
type manifestData struct {
digest digest
layers []digest
loaded bool
loadErr error
loadLock sync.Mutex
}
type manifestsData map[digest]*manifestData
 
var manifests manifestsData = make(map[digest]*manifestData)
var manifestsLock sync.Mutex
 
func deserializeManifest(data []byte) (distribution.Manifest, error) {
var versioned manifest.Versioned
if err := json.Unmarshal(data, &versioned); err != nil {
return nil, err
}
switch versioned.SchemaVersion {
case 1:
var sm schema1.SignedManifest
err := json.Unmarshal(data, &sm)
return sm, err
case 2:
// This can be an image manifest or a manifest list
switch versioned.MediaType {
case schema2.MediaTypeManifest:
var m schema2.DeserializedManifest
err := json.Unmarshal(data, &m)
return m, err
case manifestlist.MediaTypeManifestList:
var m manifestlist.DeserializedManifestList
err := json.Unmarshal(data, &m)
return m, err
default:
return nil, distribution.ErrManifestVerification{fmt.Errorf("unrecognized manifest content type %s", versioned.MediaType)}
}
}
return nil, fmt.Errorf("unrecognized manifest schema version %d", versioned.SchemaVersion)
}
func (m *manifestData) path() string {
return filepath.Join("blobs", m.digest.scopedPath(), "data")
}
func (m *manifestData) load(blobs blobsData) error {
logrus.Println("MANIFEST:", m.path(), ": loading...")
data, err := currentStorage.Read(m.path(), blobs.etag(m.digest))
if err != nil {
return err
}
manifest, err := deserializeManifest(data)
if err != nil {
return err
}
for _, reference := range manifest.References() {
digest, err := newDigestFromReference([]byte(reference.Digest))
if err != nil {
return err
}
m.layers = append(m.layers, digest)
}
return nil
}
func (m manifestsData) get(digest digest, blobs blobsData) (*manifestData, error) {
manifestsLock.Lock()
manifest := m[digest]
@@ -96,14 +20,5 @@ func (m manifestsData) get(digest digest, blobs blobsData) (*manifestData, error
}
manifestsLock.Unlock()
 
if !manifest.loaded {
manifest.loadLock.Lock()
defer manifest.loadLock.Unlock()
if !manifest.loaded {
manifest.loadErr = manifest.load(blobs)
manifest.loaded = true
}
}
return manifest, manifest.loadErr
return manifest, manifest.ensureLoaded(blobs)
}
repositories.go
+ 9
− 377
View file @ ecc8ad8c
@@ -5,246 +5,18 @@ import (
"fmt"
"io"
"os"
"path/filepath"
"strings"
"sync"
 
"github.com/Sirupsen/logrus"
"github.com/dustin/go-humanize"
)
 
type repositoryData struct {
name string
layers map[digest]int
manifests map[digest]int
manifestSignatures map[digest][]digest
tags map[string]*tagData
uploads []string
lock sync.Mutex
}
type repositoriesData map[string]*repositoryData
var repositoriesLock sync.Mutex
var parallelRepositoryWalk = flag.Bool("parallel-repository-walk", true, "Allow to use parallel repository walker")
var repositoryCsvOutput = flag.String("repository-csv-output", "repositories.csv", "File to which CSV will be written with all metrics")
 
func newRepositoryData(name string) *repositoryData {
return &repositoryData{
name: name,
layers: make(map[digest]int),
manifests: make(map[digest]int),
manifestSignatures: make(map[digest][]digest),
tags: make(map[string]*tagData),
}
}
func (r *repositoryData) layerLinkPath(layer digest) string {
return filepath.Join("repositories", r.name, "_layers", layer.path(), "link")
}
func (r *repositoryData) manifestRevisionPath(revision digest) string {
return filepath.Join("repositories", r.name, "_manifests", "revisions", revision.path(), "link")
}
func (r *repositoryData) manifestRevisionSignaturePath(revision, signature digest) string {
return filepath.Join("repositories", r.name, "_manifests", "revisions", revision.path(), "signatures", signature.path(), "link")
}
func (r *repositoryData) uploadPath(upload string) string {
return filepath.Join("repositories", r.name, "_uploads", upload, "link")
}
func (r *repositoryData) tag(name string) *tagData {
r.lock.Lock()
defer r.lock.Unlock()
t := r.tags[name]
if t == nil {
t = &tagData{
repository: r,
name: name,
}
r.tags[name] = t
}
return t
}
func (r *repositoryData) markManifest(revision digest) error {
r.lock.Lock()
defer r.lock.Unlock()
r.manifests[revision]++
return nil
}
func (r *repositoryData) markManifestLayers(blobs blobsData, revision digest) error {
err := blobs.mark(revision)
if err != nil {
return err
}
manifest, err := manifests.get(revision, blobs)
if err != nil {
return err
}
r.lock.Lock()
defer r.lock.Unlock()
for _, layer := range manifest.layers {
_, ok := r.layers[layer]
if !ok {
return fmt.Errorf("layer %s not found reference from manifest %s", layer, revision)
}
r.layers[layer]++
}
return nil
}
func (r *repositoryData) markManifestSignatures(blobs blobsData, revision digest, signatures []digest) error {
if r.manifests[revision] == 0 {
return nil
}
for _, signature := range signatures {
blobs.mark(signature)
}
return nil
}
func (r *repositoryData) sweepManifestSignatures(revision digest, signatures []digest) error {
if r.manifests[revision] > 0 {
return nil
}
for _, signature := range signatures {
err := deleteFile(r.manifestRevisionSignaturePath(revision, signature), digestReferenceSize)
if err != nil {
return err
}
}
return nil
}
func (r *repositoryData) markLayer(blobs blobsData, revision digest) error {
return blobs.mark(revision)
}
func (r *repositoryData) mark(blobs blobsData) error {
for name, t := range r.tags {
err := t.mark(blobs)
if err != nil {
if *softErrors {
logrus.Errorln("MARK:", r.name, "TAG:", name, "ERROR:", err)
continue
}
return err
}
}
for revision, used := range r.manifests {
if used == 0 {
continue
}
err := r.markManifestLayers(blobs, revision)
if err != nil {
if *softErrors {
logrus.Errorln("MARK:", r.name, "MANIFEST:", revision, "ERROR:", err)
continue
}
return err
}
}
for revision, signatures := range r.manifestSignatures {
err := r.markManifestSignatures(blobs, revision, signatures)
if err != nil {
if *softErrors {
logrus.Errorln("MARK:", r.name, "MANIFEST SIGNATURE:", revision, "ERROR:", err)
continue
}
return err
}
}
for digest, used := range r.layers {
if used == 0 {
continue
}
err := r.markLayer(blobs, digest)
if err != nil {
if *softErrors {
logrus.Errorln("MARK:", r.name, "LAYER:", digest, "ERROR:", err)
continue
}
return err
}
}
return nil
}
func (r *repositoryData) sweep() error {
for name, t := range r.tags {
err := t.sweep()
if err != nil {
if *softErrors {
logrus.Errorln("SWEEP:", r.name, "TAG:", name, "ERROR:", err)
continue
}
return err
}
}
for revision, used := range r.manifests {
if used > 0 {
continue
}
err := deleteFile(r.manifestRevisionPath(revision), digestReferenceSize)
if err != nil {
if *softErrors {
logrus.Errorln("SWEEP:", r.name, "MANIFEST:", revision, "ERROR:", err)
continue
}
return err
}
}
for revision, signatures := range r.manifestSignatures {
err := r.sweepManifestSignatures(revision, signatures)
if err != nil {
if *softErrors {
logrus.Errorln("MARK:", r.name, "MANIFEST SIGNATURES:", revision, "ERROR:", err)
continue
}
return err
}
}
for digest, used := range r.layers {
if used > 0 {
continue
}
err := deleteFile(r.layerLinkPath(digest), digestReferenceSize)
if err != nil {
if *softErrors {
logrus.Errorln("MARK:", r.name, "LAYER:", digest, "ERROR:", err)
continue
}
return err
}
}
type repositoriesData map[string]*repositoryData
 
return nil
}
var repositoriesLock sync.Mutex
 
func (r repositoriesData) get(path []string) *repositoryData {
repositoryName := strings.Join(path, "/")
@@ -261,146 +33,6 @@ func (r repositoriesData) get(path []string) *repositoryData {
return repository
}
 
func (r *repositoryData) addLayer(args []string, info fileInfo) error {
// /test/_layers/sha256/579c7fc9b0d60a19706cd6c1573fec9a28fa758bfe1ece86a1e5c68ad6f4e9d1/link
link, err := analyzeLink(args)
if err != nil {
return err
}
err = verifyLink(link, r.layerLinkPath(link), info.etag)
if err != nil {
return err
}
r.lock.Lock()
defer r.lock.Unlock()
r.layers[link] = 0
return nil
}
func (r *repositoryData) addManifestRevision(args []string, info fileInfo) error {
// /test2/_manifests/revisions/sha256/708519982eae159899e908639f5fa22d23d247ad923f6e6ad6128894c5d497a0/link
link, err := analyzeLink(args)
if err == nil {
err = verifyLink(link, r.manifestRevisionPath(link), info.etag)
if err != nil {
return err
}
r.lock.Lock()
defer r.lock.Unlock()
r.manifests[link] = 0
return nil
}
link, signature, err := analyzeLinkSignature(args)
if err == nil {
err = verifyLink(signature, r.manifestRevisionSignaturePath(link, signature), info.etag)
if err != nil {
return err
}
r.lock.Lock()
defer r.lock.Unlock()
r.manifestSignatures[link] = append(r.manifestSignatures[link], signature)
return nil
}
return err
}
func (r *repositoryData) addTag(args []string, info fileInfo) error {
//INFO[0000] /test2/_manifests/tags/latest/current/link
//INFO[0000] /test2/_manifests/tags/latest/index/sha256/af8338145978acd626bfb9e863fa446bebfc9f2660bee1af99ed29efc48d73b4/link
tag := r.tag(args[0])
if args[1] == "current" {
return tag.setCurrent(info)
} else if args[1] == "index" {
return tag.addVersion(args[2:], info)
} else {
return fmt.Errorf("undefined manifest tag type: %v", args[1])
}
}
func (r *repositoryData) addManifest(args []string, info fileInfo) error {
//INFO[0000] /test2/_manifests/revisions/sha256/708519982eae159899e908639f5fa22d23d247ad923f6e6ad6128894c5d497a0/link
//INFO[0000] /test2/_manifests/revisions/sha256/af8338145978acd626bfb9e863fa446bebfc9f2660bee1af99ed29efc48d73b4/link
//INFO[0000] /test2/_manifests/tags/latest/current/link
//INFO[0000] /test2/_manifests/tags/latest/index/sha256/af8338145978acd626bfb9e863fa446bebfc9f2660bee1af99ed29efc48d73b4/link
//INFO[0000] /test2/_manifests/tags/latest2/current/link
//INFO[0000] /test2/_manifests/tags/latest2/index/sha256/708519982eae159899e908639f5fa22d23d247ad923f6e6ad6128894c5d497a0/link
if args[0] == "revisions" {
return r.addManifestRevision(args[1:], info)
} else if args[0] == "tags" {
return r.addTag(args[1:], info)
} else {
return fmt.Errorf("undefined manifest type: %v", args[0])
}
}
func (r *repositoryData) addUpload(args []string, info fileInfo) error {
// /test/_uploads/f82d2b61-f130-4be5-b4f6-92cb18c7cf89/startedat
// /test/_uploads/f82d2b61-f130-4be5-b4f6-92cb18c7cf89/hashstates/sha256/0
if len(args) < 1 {
return fmt.Errorf("invalid args for uploads: %v", args)
}
r.lock.Lock()
defer r.lock.Unlock()
r.uploads = append(r.uploads, strings.Join(args, "/"))
return nil
}
func (r *repositoryData) info(blobs blobsData, stream io.WriteCloser) {
var layersUsed, layersUnused int
var manifestsUsed, manifestsUnused int
var tagsVersions int
var layersUsedSize, layersUnusedSize int64
for digest, used := range r.layers {
if used > 0 {
layersUsed++
layersUsedSize += blobs.size(digest)
} else {
layersUnused++
layersUnusedSize += blobs.size(digest)
}
}
for _, used := range r.manifests {
if used > 0 {
manifestsUsed++
} else {
manifestsUnused++
}
}
for _, tag := range r.tags {
tagsVersions += len(tag.versions)
}
logrus.Println("REPOSITORY INFO:", r.name, ":",
"Tags/Versions:", len(r.tags), "/", tagsVersions,
"Manifests/Unused:", manifestsUsed, "/", manifestsUnused,
"Layers/Unused:", layersUsed, "/", layersUnused,
"Data/Unused:", humanize.Bytes(uint64(layersUsedSize)), "/", humanize.Bytes(uint64(layersUnusedSize)))
if stream != nil {
fmt.Fprintf(stream, "%s,%d,%d,%d,%d,%d,%d,%s,%s,%d,%d\n",
r.name, len(r.tags), tagsVersions,
manifestsUsed, manifestsUnused,
layersUsed, layersUnused,
humanize.Bytes(uint64(layersUsedSize)), humanize.Bytes(uint64(layersUnusedSize)),
layersUsedSize/1024/1024, layersUnusedSize/1024/1024)
}
}
func (r repositoriesData) process(segments []string, info fileInfo) error {
for idx := 0; idx < len(segments)-1; idx++ {
repository := segments[0:idx]
@@ -421,10 +53,10 @@ func (r repositoriesData) process(segments []string, info fileInfo) error {
return fmt.Errorf("unparseable path: %v", segments)
}
 
func (r repositoriesData) walkPath(walkPath string, jg *jobsGroup) error {
func (r repositoriesData) walkPath(walkPath string, jg *jobGroup) error {
logrus.Infoln("REPOSITORIES DIR:", walkPath)
return currentStorage.Walk(walkPath, "repositories", func(path string, info fileInfo, err error) error {
jg.Dispatch(func() error {
jg.dispatch(func() error {
err = r.process(strings.Split(path, "/"), info)
if err != nil {
if err != nil {
@@ -462,7 +94,7 @@ func (r repositoriesData) walk() error {
}
}
 
return jg.Finish()
return jg.finish()
}
 
func (r repositoriesData) mark(blobs blobsData) error {
@@ -470,12 +102,12 @@ func (r repositoriesData) mark(blobs blobsData) error {
 
for _, repository_ := range r {
repository := repository_
jg.Dispatch(func() error {
jg.dispatch(func() error {
return repository.mark(blobs)
})
}
 
err := jg.Finish()
err := jg.finish()
if err != nil {
return err
}
@@ -487,12 +119,12 @@ func (r repositoriesData) sweep() error {
 
for _, repository_ := range r {
repository := repository_
jg.Dispatch(func() error {
jg.dispatch(func() error {
return repository.sweep()
})
}
 
err := jg.Finish()
err := jg.finish()
if err != nil {
return err
}
repository.go 0 → 100644
+ 378
− 0
View file @ ecc8ad8c
package main
import (
"fmt"
"io"
"path/filepath"
"strings"
"sync"
"github.com/Sirupsen/logrus"
"github.com/dustin/go-humanize"
)
type repositoryData struct {
name string
layers map[digest]int
manifests map[digest]int
manifestSignatures map[digest][]digest
tags map[string]*tagData
uploads []string
lock sync.Mutex
}
func (r *repositoryData) layerLinkPath(layer digest) string {
return filepath.Join("repositories", r.name, "_layers", layer.path(), "link")
}
func (r *repositoryData) manifestRevisionPath(revision digest) string {
return filepath.Join("repositories", r.name, "_manifests", "revisions", revision.path(), "link")
}
func (r *repositoryData) manifestRevisionSignaturePath(revision, signature digest) string {
return filepath.Join("repositories", r.name, "_manifests", "revisions", revision.path(), "signatures", signature.path(), "link")
}
func (r *repositoryData) uploadPath(upload string) string {
return filepath.Join("repositories", r.name, "_uploads", upload, "link")
}
func (r *repositoryData) tag(name string) *tagData {
r.lock.Lock()
defer r.lock.Unlock()
t := r.tags[name]
if t == nil {
t = &tagData{
repository: r,
name: name,
}
r.tags[name] = t
}
return t
}
func (r *repositoryData) markManifest(revision digest) error {
r.lock.Lock()
defer r.lock.Unlock()
r.manifests[revision]++
return nil
}
func (r *repositoryData) markManifestLayers(blobs blobsData, revision digest) error {
err := blobs.mark(revision)
if err != nil {
return err
}
manifest, err := manifests.get(revision, blobs)
if err != nil {
return err
}
r.lock.Lock()
defer r.lock.Unlock()
for _, layer := range manifest.layers {
_, ok := r.layers[layer]
if !ok {
return fmt.Errorf("layer %s not found reference from manifest %s", layer, revision)
}
r.layers[layer]++
}
return nil
}
func (r *repositoryData) markManifestSignatures(blobs blobsData, revision digest, signatures []digest) error {
if r.manifests[revision] == 0 {
return nil
}
for _, signature := range signatures {
blobs.mark(signature)
}
return nil
}
func (r *repositoryData) sweepManifestSignatures(revision digest, signatures []digest) error {
if r.manifests[revision] > 0 {
return nil
}
for _, signature := range signatures {
err := deleteFile(r.manifestRevisionSignaturePath(revision, signature), digestReferenceSize)
if err != nil {
return err
}
}
return nil
}
func (r *repositoryData) markLayer(blobs blobsData, revision digest) error {
return blobs.mark(revision)
}
func (r *repositoryData) mark(blobs blobsData) error {
for name, t := range r.tags {
err := t.mark(blobs)
if err != nil {
if *softErrors {
logrus.Errorln("MARK:", r.name, "TAG:", name, "ERROR:", err)
continue
}
return err
}
}
for revision, used := range r.manifests {
if used == 0 {
continue
}
err := r.markManifestLayers(blobs, revision)
if err != nil {
if *softErrors {
logrus.Errorln("MARK:", r.name, "MANIFEST:", revision, "ERROR:", err)
continue
}
return err
}
}
for revision, signatures := range r.manifestSignatures {
err := r.markManifestSignatures(blobs, revision, signatures)
if err != nil {
if *softErrors {
logrus.Errorln("MARK:", r.name, "MANIFEST SIGNATURE:", revision, "ERROR:", err)
continue
}
return err
}
}
for digest, used := range r.layers {
if used == 0 {
continue
}
err := r.markLayer(blobs, digest)
if err != nil {
if *softErrors {
logrus.Errorln("MARK:", r.name, "LAYER:", digest, "ERROR:", err)
continue
}
return err
}
}
return nil
}
func (r *repositoryData) sweep() error {
for name, t := range r.tags {
err := t.sweep()
if err != nil {
if *softErrors {
logrus.Errorln("SWEEP:", r.name, "TAG:", name, "ERROR:", err)
continue
}
return err
}
}
for revision, used := range r.manifests {
if used > 0 {
continue
}
err := deleteFile(r.manifestRevisionPath(revision), digestReferenceSize)
if err != nil {
if *softErrors {
logrus.Errorln("SWEEP:", r.name, "MANIFEST:", revision, "ERROR:", err)
continue
}
return err
}
}
for revision, signatures := range r.manifestSignatures {
err := r.sweepManifestSignatures(revision, signatures)
if err != nil {
if *softErrors {
logrus.Errorln("MARK:", r.name, "MANIFEST SIGNATURES:", revision, "ERROR:", err)
continue
}
return err
}
}
for digest, used := range r.layers {
if used > 0 {
continue
}
err := deleteFile(r.layerLinkPath(digest), digestReferenceSize)
if err != nil {
if *softErrors {
logrus.Errorln("MARK:", r.name, "LAYER:", digest, "ERROR:", err)
continue
}
return err
}
}
return nil
}
func (r *repositoryData) addLayer(args []string, info fileInfo) error {
// /test/_layers/sha256/579c7fc9b0d60a19706cd6c1573fec9a28fa758bfe1ece86a1e5c68ad6f4e9d1/link
link, err := analyzeLink(args)
if err != nil {
return err
}
err = verifyLink(link, r.layerLinkPath(link), info.etag)
if err != nil {
return err
}
r.lock.Lock()
defer r.lock.Unlock()
r.layers[link] = 0
return nil
}
func (r *repositoryData) addManifestRevision(args []string, info fileInfo) error {
// /test2/_manifests/revisions/sha256/708519982eae159899e908639f5fa22d23d247ad923f6e6ad6128894c5d497a0/link
link, err := analyzeLink(args)
if err == nil {
err = verifyLink(link, r.manifestRevisionPath(link), info.etag)
if err != nil {
return err
}
r.lock.Lock()
defer r.lock.Unlock()
r.manifests[link] = 0
return nil
}
link, signature, err := analyzeLinkSignature(args)
if err == nil {
err = verifyLink(signature, r.manifestRevisionSignaturePath(link, signature), info.etag)
if err != nil {
return err
}
r.lock.Lock()
defer r.lock.Unlock()
r.manifestSignatures[link] = append(r.manifestSignatures[link], signature)
return nil
}
return err
}
func (r *repositoryData) addTag(args []string, info fileInfo) error {
//INFO[0000] /test2/_manifests/tags/latest/current/link
//INFO[0000] /test2/_manifests/tags/latest/index/sha256/af8338145978acd626bfb9e863fa446bebfc9f2660bee1af99ed29efc48d73b4/link
tag := r.tag(args[0])
if args[1] == "current" {
return tag.setCurrent(info)
} else if args[1] == "index" {
return tag.addVersion(args[2:], info)
} else {
return fmt.Errorf("undefined manifest tag type: %v", args[1])
}
}
func (r *repositoryData) addManifest(args []string, info fileInfo) error {
//INFO[0000] /test2/_manifests/revisions/sha256/708519982eae159899e908639f5fa22d23d247ad923f6e6ad6128894c5d497a0/link
//INFO[0000] /test2/_manifests/revisions/sha256/af8338145978acd626bfb9e863fa446bebfc9f2660bee1af99ed29efc48d73b4/link
//INFO[0000] /test2/_manifests/tags/latest/current/link
//INFO[0000] /test2/_manifests/tags/latest/index/sha256/af8338145978acd626bfb9e863fa446bebfc9f2660bee1af99ed29efc48d73b4/link
//INFO[0000] /test2/_manifests/tags/latest2/current/link
//INFO[0000] /test2/_manifests/tags/latest2/index/sha256/708519982eae159899e908639f5fa22d23d247ad923f6e6ad6128894c5d497a0/link
if args[0] == "revisions" {
return r.addManifestRevision(args[1:], info)
} else if args[0] == "tags" {
return r.addTag(args[1:], info)
} else {
return fmt.Errorf("undefined manifest type: %v", args[0])
}
}
func (r *repositoryData) addUpload(args []string, info fileInfo) error {
// /test/_uploads/f82d2b61-f130-4be5-b4f6-92cb18c7cf89/startedat
// /test/_uploads/f82d2b61-f130-4be5-b4f6-92cb18c7cf89/hashstates/sha256/0
if len(args) < 1 {
return fmt.Errorf("invalid args for uploads: %v", args)
}
r.lock.Lock()
defer r.lock.Unlock()
r.uploads = append(r.uploads, strings.Join(args, "/"))
return nil
}
func (r *repositoryData) info(blobs blobsData, stream io.WriteCloser) {
var layersUsed, layersUnused int
var manifestsUsed, manifestsUnused int
var tagsVersions int
var layersUsedSize, layersUnusedSize int64
for digest, used := range r.layers {
if used > 0 {
layersUsed++
layersUsedSize += blobs.size(digest)
} else {
layersUnused++
layersUnusedSize += blobs.size(digest)
}
}
for _, used := range r.manifests {
if used > 0 {
manifestsUsed++
} else {
manifestsUnused++
}
}
for _, tag := range r.tags {
tagsVersions += len(tag.versions)
}
logrus.Println("REPOSITORY INFO:", r.name, ":",
"Tags/Versions:", len(r.tags), "/", tagsVersions,
"Manifests/Unused:", manifestsUsed, "/", manifestsUnused,
"Layers/Unused:", layersUsed, "/", layersUnused,
"Data/Unused:", humanize.Bytes(uint64(layersUsedSize)), "/", humanize.Bytes(uint64(layersUnusedSize)))
if stream != nil {
fmt.Fprintf(stream, "%s,%d,%d,%d,%d,%d,%d,%s,%s,%d,%d\n",
r.name, len(r.tags), tagsVersions,
manifestsUsed, manifestsUnused,
layersUsed, layersUnused,
humanize.Bytes(uint64(layersUsedSize)), humanize.Bytes(uint64(layersUnusedSize)),
layersUsedSize/1024/1024, layersUnusedSize/1024/1024)
}
}
func newRepositoryData(name string) *repositoryData {
return &repositoryData{
name: name,
layers: make(map[digest]int),
manifests: make(map[digest]int),
manifestSignatures: make(map[digest][]digest),
tags: make(map[string]*tagData),
}
}
storage.go
+ 2
− 2
View file @ ecc8ad8c
@@ -34,7 +34,7 @@ func parallelWalk(rootPath string, fn func(string) error) error {
return nil
}
 
pwg.Dispatch(func() error {
pwg.dispatch(func() error {
walkPath := filepath.Join(rootPath, listPath)
return fn(walkPath)
})
@@ -44,5 +44,5 @@ func parallelWalk(rootPath string, fn func(string) error) error {
return err
}
 
return pwg.Finish()
return pwg.finish()
}
tags.go → tag.go
+ 0
− 0
View file @ ecc8ad8c
File moved
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment