storage.go

Documentation: helm.sh/helm/v3/pkg/storage

     1  /*
     2  Copyright The Helm Authors.
     3  
     4  Licensed under the Apache License, Version 2.0 (the "License");
     5  you may not use this file except in compliance with the License.
     6  You may obtain a copy of the License at
     7  
     8      http://www.apache.org/licenses/LICENSE-2.0
     9  
    10  Unless required by applicable law or agreed to in writing, software
    11  distributed under the License is distributed on an "AS IS" BASIS,
    12  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
    13  See the License for the specific language governing permissions and
    14  limitations under the License.
    15  */
    16  
    17  package storage // import "helm.sh/helm/v3/pkg/storage"
    18  
    19  import (
    20  	"fmt"
    21  	"strings"
    22  
    23  	"github.com/pkg/errors"
    24  
    25  	rspb "helm.sh/helm/v3/pkg/release"
    26  	relutil "helm.sh/helm/v3/pkg/releaseutil"
    27  	"helm.sh/helm/v3/pkg/storage/driver"
    28  )
    29  
    30  // HelmStorageType is the type field of the Kubernetes storage object which stores the Helm release
    31  // version. It is modified slightly replacing the '/': sh.helm/release.v1
    32  // Note: The version 'v1' is incremented if the release object metadata is
    33  // modified between major releases.
    34  // This constant is used as a prefix for the Kubernetes storage object name.
    35  const HelmStorageType = "sh.helm.release.v1"
    36  
    37  // Storage represents a storage engine for a Release.
    38  type Storage struct {
    39  	driver.Driver
    40  
    41  	// MaxHistory specifies the maximum number of historical releases that will
    42  	// be retained, including the most recent release. Values of 0 or less are
    43  	// ignored (meaning no limits are imposed).
    44  	MaxHistory int
    45  
    46  	Log func(string, ...interface{})
    47  }
    48  
    49  // Get retrieves the release from storage. An error is returned
    50  // if the storage driver failed to fetch the release, or the
    51  // release identified by the key, version pair does not exist.
    52  func (s *Storage) Get(name string, version int) (*rspb.Release, error) {
    53  	s.Log("getting release %q", makeKey(name, version))
    54  	return s.Driver.Get(makeKey(name, version))
    55  }
    56  
    57  // Create creates a new storage entry holding the release. An
    58  // error is returned if the storage driver fails to store the
    59  // release, or a release with an identical key already exists.
    60  func (s *Storage) Create(rls *rspb.Release) error {
    61  	s.Log("creating release %q", makeKey(rls.Name, rls.Version))
    62  	if s.MaxHistory > 0 {
    63  		// Want to make space for one more release.
    64  		if err := s.removeLeastRecent(rls.Name, s.MaxHistory-1); err != nil &&
    65  			!errors.Is(err, driver.ErrReleaseNotFound) {
    66  			return err
    67  		}
    68  	}
    69  	return s.Driver.Create(makeKey(rls.Name, rls.Version), rls)
    70  }
    71  
    72  // Update updates the release in storage. An error is returned if the
    73  // storage backend fails to update the release or if the release
    74  // does not exist.
    75  func (s *Storage) Update(rls *rspb.Release) error {
    76  	s.Log("updating release %q", makeKey(rls.Name, rls.Version))
    77  	return s.Driver.Update(makeKey(rls.Name, rls.Version), rls)
    78  }
    79  
    80  // Delete deletes the release from storage. An error is returned if
    81  // the storage backend fails to delete the release or if the release
    82  // does not exist.
    83  func (s *Storage) Delete(name string, version int) (*rspb.Release, error) {
    84  	s.Log("deleting release %q", makeKey(name, version))
    85  	return s.Driver.Delete(makeKey(name, version))
    86  }
    87  
    88  // ListReleases returns all releases from storage. An error is returned if the
    89  // storage backend fails to retrieve the releases.
    90  func (s *Storage) ListReleases() ([]*rspb.Release, error) {
    91  	s.Log("listing all releases in storage")
    92  	return s.Driver.List(func(_ *rspb.Release) bool { return true })
    93  }
    94  
    95  // ListUninstalled returns all releases with Status == UNINSTALLED. An error is returned
    96  // if the storage backend fails to retrieve the releases.
    97  func (s *Storage) ListUninstalled() ([]*rspb.Release, error) {
    98  	s.Log("listing uninstalled releases in storage")
    99  	return s.Driver.List(func(rls *rspb.Release) bool {
   100  		return relutil.StatusFilter(rspb.StatusUninstalled).Check(rls)
   101  	})
   102  }
   103  
   104  // ListDeployed returns all releases with Status == DEPLOYED. An error is returned
   105  // if the storage backend fails to retrieve the releases.
   106  func (s *Storage) ListDeployed() ([]*rspb.Release, error) {
   107  	s.Log("listing all deployed releases in storage")
   108  	return s.Driver.List(func(rls *rspb.Release) bool {
   109  		return relutil.StatusFilter(rspb.StatusDeployed).Check(rls)
   110  	})
   111  }
   112  
   113  // Deployed returns the last deployed release with the provided release name, or
   114  // returns ErrReleaseNotFound if not found.
   115  func (s *Storage) Deployed(name string) (*rspb.Release, error) {
   116  	ls, err := s.DeployedAll(name)
   117  	if err != nil {
   118  		return nil, err
   119  	}
   120  
   121  	if len(ls) == 0 {
   122  		return nil, driver.NewErrNoDeployedReleases(name)
   123  	}
   124  
   125  	// If executed concurrently, Helm's database gets corrupted
   126  	// and multiple releases are DEPLOYED. Take the latest.
   127  	relutil.Reverse(ls, relutil.SortByRevision)
   128  
   129  	return ls[0], nil
   130  }
   131  
   132  // DeployedAll returns all deployed releases with the provided name, or
   133  // returns ErrReleaseNotFound if not found.
   134  func (s *Storage) DeployedAll(name string) ([]*rspb.Release, error) {
   135  	s.Log("getting deployed releases from %q history", name)
   136  
   137  	ls, err := s.Driver.Query(map[string]string{
   138  		"name":   name,
   139  		"owner":  "helm",
   140  		"status": "deployed",
   141  	})
   142  	if err == nil {
   143  		return ls, nil
   144  	}
   145  	if strings.Contains(err.Error(), "not found") {
   146  		return nil, driver.NewErrNoDeployedReleases(name)
   147  	}
   148  	return nil, err
   149  }
   150  
   151  // History returns the revision history for the release with the provided name, or
   152  // returns ErrReleaseNotFound if no such release name exists.
   153  func (s *Storage) History(name string) ([]*rspb.Release, error) {
   154  	s.Log("getting release history for %q", name)
   155  
   156  	return s.Driver.Query(map[string]string{"name": name, "owner": "helm"})
   157  }
   158  
   159  // removeLeastRecent removes items from history until the length number of releases
   160  // does not exceed max.
   161  //
   162  // We allow max to be set explicitly so that calling functions can "make space"
   163  // for the new records they are going to write.
   164  func (s *Storage) removeLeastRecent(name string, max int) error {
   165  	if max < 0 {
   166  		return nil
   167  	}
   168  	h, err := s.History(name)
   169  	if err != nil {
   170  		return err
   171  	}
   172  	if len(h) <= max {
   173  		return nil
   174  	}
   175  
   176  	// We want oldest to newest
   177  	relutil.SortByRevision(h)
   178  
   179  	lastDeployed, err := s.Deployed(name)
   180  	if err != nil && !errors.Is(err, driver.ErrNoDeployedReleases) {
   181  		return err
   182  	}
   183  
   184  	var toDelete []*rspb.Release
   185  	for _, rel := range h {
   186  		// once we have enough releases to delete to reach the max, stop
   187  		if len(h)-len(toDelete) == max {
   188  			break
   189  		}
   190  		if lastDeployed != nil {
   191  			if rel.Version != lastDeployed.Version {
   192  				toDelete = append(toDelete, rel)
   193  			}
   194  		} else {
   195  			toDelete = append(toDelete, rel)
   196  		}
   197  	}
   198  
   199  	// Delete as many as possible. In the case of API throughput limitations,
   200  	// multiple invocations of this function will eventually delete them all.
   201  	errs := []error{}
   202  	for _, rel := range toDelete {
   203  		err = s.deleteReleaseVersion(name, rel.Version)
   204  		if err != nil {
   205  			errs = append(errs, err)
   206  		}
   207  	}
   208  
   209  	s.Log("Pruned %d record(s) from %s with %d error(s)", len(toDelete), name, len(errs))
   210  	switch c := len(errs); c {
   211  	case 0:
   212  		return nil
   213  	case 1:
   214  		return errs[0]
   215  	default:
   216  		return errors.Errorf("encountered %d deletion errors. First is: %s", c, errs[0])
   217  	}
   218  }
   219  
   220  func (s *Storage) deleteReleaseVersion(name string, version int) error {
   221  	key := makeKey(name, version)
   222  	_, err := s.Delete(name, version)
   223  	if err != nil {
   224  		s.Log("error pruning %s from release history: %s", key, err)
   225  		return err
   226  	}
   227  	return nil
   228  }
   229  
   230  // Last fetches the last revision of the named release.
   231  func (s *Storage) Last(name string) (*rspb.Release, error) {
   232  	s.Log("getting last revision of %q", name)
   233  	h, err := s.History(name)
   234  	if err != nil {
   235  		return nil, err
   236  	}
   237  	if len(h) == 0 {
   238  		return nil, errors.Errorf("no revision for release %q", name)
   239  	}
   240  
   241  	relutil.Reverse(h, relutil.SortByRevision)
   242  	return h[0], nil
   243  }
   244  
   245  // makeKey concatenates the Kubernetes storage object type, a release name and version
   246  // into a string with format:```<helm_storage_type>.<release_name>.v<release_version>```.
   247  // The storage type is prepended to keep name uniqueness between different
   248  // release storage types. An example of clash when not using the type:
   249  // https://github.com/helm/helm/issues/6435.
   250  // This key is used to uniquely identify storage objects.
   251  func makeKey(rlsname string, version int) string {
   252  	return fmt.Sprintf("%s.%s.v%d", HelmStorageType, rlsname, version)
   253  }
   254  
   255  // Init initializes a new storage backend with the driver d.
   256  // If d is nil, the default in-memory driver is used.
   257  func Init(d driver.Driver) *Storage {
   258  	// default driver is in memory
   259  	if d == nil {
   260  		d = driver.NewMemory()
   261  	}
   262  	return &Storage{
   263  		Driver: d,
   264  		Log:    func(_ string, _ ...interface{}) {},
   265  	}
   266  }
   267
View as plain text