func UploadSpecificKeys(ctx context.Context, bundle *Bundle, getKeys func() ([]string, error), opts ...Option) (err error) {
	defer func(t0 time.Time) {
		if bundle.MetricsEnabled() {
			bundle.m.Usage.UsedAll(t0, "UploadSpecificKeys")(err)
		}
	}(time.Now())
	err = implUpload(ctx, bundle, defaultBundleEntriesPerFile, getKeys, opts...)
	return
}
UploadSpecificKeys uploads some specified keys (files) within a bundle's consumable store
package core
import (
	"context"
	"fmt"
	"time"
	context2 "github.com/oneconcern/datamon/pkg/context"
	"github.com/oneconcern/datamon/pkg/metrics"
	"github.com/oneconcern/datamon/pkg/dlogger"
	"go.uber.org/zap"
	"github.com/segmentio/ksuid"
	"github.com/oneconcern/datamon/pkg/core/status"
	"github.com/oneconcern/datamon/pkg/model"
	"github.com/oneconcern/datamon/pkg/storage"
)
type Bundle struct {
	RepoID                      string
	BundleID                    string
	ConsumableStore             storage.Store
	contextStores               context2.Stores
	BundleDescriptor            model.BundleDescriptor
	BundleEntries               []model.BundleEntry
	l                           *zap.Logger
	SkipOnError                 bool  
	Retry                       bool  
	concurrentFileUploads       int
	concurrentFileDownloads     int
	concurrentFilelistDownloads int
	metrics.Enable
	m *M
}
func (b *Bundle) setBundleID(id string) {
	b.BundleID = id
	b.BundleDescriptor.ID = id
}
func (b *Bundle) InitializeBundleID() error {
	id, err := ksuid.NewRandom()
	if err != nil {
		return err
	}
	b.setBundleID(id.String())
	return nil
}
func (b *Bundle) GetBundleEntries() []model.BundleEntry {
	return b.BundleEntries
}
func (b *Bundle) BlobStore() storage.Store {
	return getBlobStore(b.contextStores)
}
func getBlobStore(stores context2.Stores) storage.Store {
	if stores == nil {
		return nil
	}
	return stores.Blob()
}
func (b *Bundle) MetaStore() storage.Store {
	return getMetaStore(b.contextStores)
}
func getMetaStore(stores context2.Stores) storage.Store {
	if stores == nil {
		return nil
	}
	return stores.Metadata()
}
func (b *Bundle) VMetaStore() storage.Store {
	return getVMetaStore(b.contextStores)
}
func getVMetaStore(stores context2.Stores) storage.Store {
	if stores == nil {
		return nil
	}
	return stores.VMetadata()
}
func (b *Bundle) WALStore() storage.Store {
	return getWALStore(b.contextStores)
}
func getWALStore(stores context2.Stores) storage.Store {
	if stores == nil {
		return nil
	}
	return stores.Wal()
}
func (b *Bundle) ReadLogStore() storage.Store {
	return getReadLogStore(b.contextStores)
}
func getReadLogStore(stores context2.Stores) storage.Store {
	if stores == nil {
		return nil
	}
	return stores.ReadLog()
}
func defaultBundle() *Bundle {
	return &Bundle{
		BundleDescriptor:            *model.NewBundleDescriptor(),
		RepoID:                      "",
		BundleID:                    "",
		ConsumableStore:             nil,
		BundleEntries:               make([]model.BundleEntry, 0, 1024),
		concurrentFileUploads:       20,
		concurrentFileDownloads:     10,
		concurrentFilelistDownloads: 10,
		l:                           dlogger.MustGetLogger("info"),
	}
}
func NewBundle(opts ...BundleOption) *Bundle {
	b := defaultBundle()
	for _, apply := range opts {
		apply(b)
	}
	if b.MetricsEnabled() {
		b.m = b.EnsureMetrics("core", &M{}).(*M)
	}
	return b
}
func Publish(ctx context.Context, bundle *Bundle) (err error) {
	defer func(t0 time.Time) {
		if bundle.MetricsEnabled() {
			bundle.m.Usage.UsedAll(t0, "Publish")(err)
		}
	}(time.Now())
	err = implPublish(ctx, bundle, defaultBundleEntriesPerFile, func(s string) (bool, error) { return true, nil })
	return
}
func PublishSelectBundleEntries(ctx context.Context, bundle *Bundle, selectionPredicate func(string) (bool, error)) (err error) {
	defer func(t0 time.Time) {
		if bundle.MetricsEnabled() {
			bundle.m.Usage.UsedAll(t0, "PublishSelectBundleEntries")(err)
		}
	}(time.Now())
	err = implPublish(ctx, bundle, defaultBundleEntriesPerFile, selectionPredicate)
	return
}
func implPublish(ctx context.Context, bundle *Bundle, entriesPerFile uint,
	selectionPredicate func(string) (bool, error)) error {
	err := implPublishMetadata(ctx, bundle, true, entriesPerFile)
	if err != nil {
		return status.ErrPublishMetadata.Wrap(err)
	}
	err = unpackDataFiles(ctx, bundle, nil, selectionPredicate)
	if err != nil {
		return status.ErrPublishMetadata.Wrap(err)
	}
	return nil
}
func PublishMetadata(ctx context.Context, bundle *Bundle) (err error) {
	defer func(t0 time.Time) {
		if bundle.MetricsEnabled() {
			bundle.m.Usage.UsedAll(t0, "PublishMetadata")(err)
		}
	}(time.Now())
	err = implPublishMetadata(ctx, bundle, true, defaultBundleEntriesPerFile)
	return
}
func DownloadMetadata(ctx context.Context, bundle *Bundle) (err error) {
	defer func(t0 time.Time) {
		if bundle.MetricsEnabled() {
			bundle.m.Usage.UsedAll(t0, "DownloadMetadata")(err)
		}
	}(time.Now())
	err = implPublishMetadata(ctx, bundle, false, defaultBundleEntriesPerFile)
	return
}
func implPublishMetadata(ctx context.Context, bundle *Bundle, publish bool, entriesPerFile uint) error {
	if bundle.BundleID == "" {
		if bundle.ConsumableStore != nil {
			if err := setBundleIDFromConsumableStore(ctx, bundle); err != nil {
				return err
			}
		} else {
			return status.ErrNoBundleIDWithConsumable
		}
	}
	if err := unpackBundleDescriptor(ctx, bundle, publish); err != nil {
		return err
	}
	if err := unpackBundleFileList(ctx, bundle, publish, entriesPerFile); err != nil {
		return err
	}
	return nil
}
func Upload(ctx context.Context, bundle *Bundle, opts ...Option) (err error) {
	defer func(t0 time.Time) {
		if bundle.MetricsEnabled() {
			bundle.m.Usage.UsedAll(t0, "Upload")(err)
		}
	}(time.Now())
	err = implUpload(ctx, bundle, defaultBundleEntriesPerFile, nil, opts...)
	return
}