wip: fix azure storage path to default to leading slash

also make sure that installations running the code that removed the
trailing slash will continue to work by detecting existing blobs in the
storage.

registry/storage/driver/azure/azure.go

@@ -28,9 +28,10 @@ const (
 )
 
 type driver struct {
-	azClient      *azureClient
-	client        *container.Client
-	rootDirectory string
+	azClient        *azureClient
+	client          *container.Client
+	rootDirectory   string
+	useLeadingSlash bool
 }
 
 type baseEmbed struct{ base.Base }
@@ -65,7 +66,59 @@ func New(params *Parameters) (*Driver, error) {
 		client:        client,
 		rootDirectory: params.RootDirectory,
 	}
-	return &Driver{baseEmbed: baseEmbed{Base: base.Base{StorageDriver: d}}}, nil
+	driver := &Driver{baseEmbed: baseEmbed{Base: base.Base{StorageDriver: d}}}
+
+	// the previous code removed the leading "/" on paths.
+	// this caused backwards incompatible upgrades between v2 and v3
+	// registries.
+	// because some of our customers already upgraded to v3, we are
+	// forced to provide a way to support both: paths that begin with "/",
+	// and paths that don't.
+
+	// paths start with a leading slach by default in v2 - to be backwards
+	// compatible we default to that and set it to false if we detect the
+	// existing storage does not use it.
+
+	// installations that upgraded from v2 to v3 (previous to this patch)
+	// might have data in / in their storage account, but will actually be
+	// using data from a path without the leading /.
+	// to make sure there is no data loss, we first check for data in the
+	// path with no leading slash, and only when we do not find data there,
+	// we default to the leading slash
+	d.useLeadingSlash = false
+
+	// we use listBlobs instead of List because List is recursive,
+	// and we only need to look at the root.
+	blobRef := d.client.NewBlobClient("/")
+	_, err = blobRef.GetProperties(context.TODO(), nil)
+	// TODO: revisit error handling?
+	if err != nil && !is404(err) {
+		return nil, err
+	}
+
+	// a fresh registry install will contain no blobs in its root dir.
+	// a registry that used the previous v3 path handling however might
+	// also return 0 blobs under "/", but actually contain blobs under "".
+	if is404(err) {
+		fmt.Println("GetProperties of empty blob returned 404")
+		d.useLeadingSlash = true
+		blobRef := d.client.NewBlobClient("//")
+		_, err := blobRef.GetProperties(context.TODO(), nil)
+		// TODO: revisit error handling?
+		if err != nil && !is404(err) {
+			return nil, err
+		}
+		// blobs, err := d.listBlobs(context.TODO(), "/")
+		// // TODO: revisit error handling?
+		// if err != nil {
+		// 	return nil, err
+		// }
+		// if len(blobs) == 0 {
+		// 	d.useLeadingSlash = false
+		// }
+	}
+
+	return driver, nil
 }
 
 // Implement the storagedriver.StorageDriver interface.
@@ -161,6 +214,7 @@ func (d *driver) Reader(ctx context.Context, path string, offset int64) (io.Read
 // at the location designated by "path" after the call to Commit.
 func (d *driver) Writer(ctx context.Context, path string, append bool) (storagedriver.FileWriter, error) {
 	blobName := d.blobName(path)
+	fmt.Println("Writer blobName:", blobName)
 	blobRef := d.client.NewBlobClient(blobName)
 
 	props, err := blobRef.GetProperties(ctx, nil)
@@ -258,6 +312,7 @@ func (d *driver) Stat(ctx context.Context, path string) (storagedriver.FileInfo,
 // List returns a list of the objects that are direct descendants of the given
 // path.
 func (d *driver) List(ctx context.Context, path string) ([]string, error) {
+	// the base storage driver errors if the given path is ""
 	if path == "/" {
 		path = ""
 	}
@@ -372,18 +427,23 @@ func (d *driver) Walk(ctx context.Context, path string, f storagedriver.WalkFn)
 // Example: direct descendants of "/" in {"/foo", "/bar/1", "/bar/2"} is
 // {"/foo", "/bar"} and direct descendants of "bar" is {"/bar/1", "/bar/2"}
 func directDescendants(blobs []string, prefix string) []string {
-	if !strings.HasPrefix(prefix, "/") { // add trailing '/'
+	if !strings.HasPrefix(prefix, "/") { // add leading '/'
 		prefix = "/" + prefix
 	}
 	if !strings.HasSuffix(prefix, "/") { // containerify the path
 		prefix += "/"
 	}
 
 	out := make(map[string]bool)
+	// fmt.Println("directDescendants: prefix is:", prefix)
 	for _, b := range blobs {
+		// fmt.Println("directDescendants: analyzing blob", b)
 		if strings.HasPrefix(b, prefix) {
+			// fmt.Println("blob", b, "has prefix", prefix)
 			rel := b[len(prefix):]
 			c := strings.Count(rel, "/")
+			// fmt.Println("rel is", rel)
+			// fmt.Println("c is", c)
 			if c == 0 {
 				out[b] = true
 			} else {
@@ -400,24 +460,15 @@ func directDescendants(blobs []string, prefix string) []string {
 }
 
 func (d *driver) listBlobs(ctx context.Context, virtPath string) ([]string, error) {
+	// we want the path to end in a "/" (not sure why), so if it's not empty
+	// and it doesn't already end in one, add it.
 	if virtPath != "" && !strings.HasSuffix(virtPath, "/") { // containerify the path
 		virtPath += "/"
 	}
 
-	// we will replace the root directory prefix before returning blob names
-	blobPrefix := d.blobName("")
-
-	// This is to cover for the cases when the rootDirectory of the driver is either "" or "/".
-	// In those cases, there is no root prefix to replace and we must actually add a "/" to all
-	// results in order to keep them as valid paths as recognized by storagedriver.PathRegexp
-	prefix := ""
-	if blobPrefix == "" {
-		prefix = "/"
-	}
-
 	out := []string{}
-
 	listPrefix := d.blobName(virtPath)
+
 	pager := d.client.NewListBlobsFlatPager(&container.ListBlobsFlatOptions{
 		Prefix: &listPrefix,
 	})
@@ -431,10 +482,20 @@ func (d *driver) listBlobs(ctx context.Context, virtPath string) ([]string, erro
 				return nil, fmt.Errorf("required blob property Name is missing while listing blobs under: %s", listPrefix)
 			}
 			name := *blob.Name
-			out = append(out, strings.Replace(name, blobPrefix, prefix, 1))
+
+			// This is to cover for the cases when the rootDirectory of the driver is either "" or "/".
+			// In those cases, there is no root prefix to replace and we must actually add a "/" to all
+			// results in order to keep them as valid paths as recognized by storagedriver.PathRegexp
+			if !strings.HasPrefix(name, "/") { // add leading '/'
+				name = "/" + name
+			}
+			out = append(out, name)
 		}
 	}
 
+	// fmt.Println("listBlob out's len is:", len(out))
+	// fmt.Println("listBlob out[0] is:", out[0])
+
 	return out, nil
 }
 
@@ -446,7 +507,23 @@ func (d *driver) blobName(path string) string {
 		return path
 	}
 
-	return strings.TrimLeft(strings.TrimRight(d.rootDirectory, "/")+path, "/")
+	// the previous code removed the leading "/" on paths.
+	// this caused backwards incompatible upgrades between v2 and v3
+	// registries.
+	// because some of our customers already upgraded to v3, we are
+	// forced to provide a way to support both: paths that begin with "/",
+	// and paths that don't.
+
+	trimmedRoot := strings.TrimRight(d.rootDirectory, "/")
+	trimmedPath := strings.TrimLeft(path, "/")
+	if d.useLeadingSlash {
+		if !strings.HasPrefix(trimmedRoot, "/") {
+			trimmedRoot = "/" + trimmedRoot
+		}
+		return trimmedRoot + "/" + trimmedPath
+	}
+
+	return trimmedRoot + trimmedPath
 }
 
 func is404(err error) bool {

registry/storage/driver/azure/azure_test.go

@@ -91,6 +91,147 @@ func randStringRunes(n int) string {
 	return string(b)
 }
 
+func writeBlob(ctx context.Context, driver storagedriver.StorageDriver, contents, path string, append_ bool) error {
+	writer, err := driver.Writer(ctx, path, append_)
+	if err != nil {
+		return fmt.Errorf("unexpected error from driver.Writer: %v", err)
+	}
+	_, err = writer.Write([]byte(contents))
+	if err != nil {
+		return fmt.Errorf("writer.Write: unexpected error: %v", err)
+	}
+
+	err = writer.Commit()
+	if err != nil {
+		return fmt.Errorf("writer.Commit: unexpected error: %v", err)
+	}
+	err = writer.Close()
+	if err != nil {
+		return fmt.Errorf("writer.Close: unexpected error: %v", err)
+	}
+	return nil
+}
+
+func TestUseTrailingSlash(t *testing.T) {
+	ctx := context.Background()
+	fmt.Println("================================== empty storage ======================================")
+	d, err := azureDriverConstructor()
+	if err != nil {
+		t.Fatalf("unexpected error creating azure driver: %v", err)
+	}
+	storageDriver := d.(*Driver)
+	drvr := storageDriver.Base.StorageDriver.(*driver)
+
+	// at this point there should be no data in the storage with or without
+	// a leading slash. this simulates a fresh install of the registry, so we
+	// expect the default useLeadingSlash to be used (true).
+	if !drvr.useLeadingSlash {
+		t.Fatal("expected a fresh install of the registry to use a leading slash on storage paths by default")
+	}
+	fmt.Println("========================================================================")
+
+	fmt.Println("================================== /docker/<etc> ======================================")
+	// create files in storage with leading slash
+	// this should detect that files starting with a /
+	// exist, which will set useLeadingSlash to true
+	contents := randStringRunes(4 * 1024 * 1024)
+	path1 := "/docker/registry/v2/repositories/test/hello/_layers/sha256/abc123/link"
+
+	defer d.Delete(ctx, path1)
+	if err := writeBlob(ctx, d, contents, path1, false); err != nil {
+		t.Fatal(err)
+	}
+
+	// create another instance of the storage driver. this is the only way
+	// to reset the value of useLeadingSlash
+	d, err = azureDriverConstructor()
+	if err != nil {
+		t.Fatalf("unexpected error creating azure driver: %v", err)
+	}
+	storageDriver = d.(*Driver)
+	drvr = storageDriver.Base.StorageDriver.(*driver)
+
+	// when files exist in a path with a leading / the registry is using
+	// existing storage and it should follow the /, useLeadingSlash should
+	// be true.
+	if !drvr.useLeadingSlash {
+		t.Error("expected an existing install with data under `/docker` to use a leading slash on storage paths by default")
+		t.Logf("sample blob name: %s", drvr.blobName("/docker/registry/v2/foo"))
+	}
+	fmt.Println("========================================================================")
+
+	fmt.Println("========================================================================")
+	fmt.Println("================================== docker/<etc> ======================================")
+	// now we'll simulate a case where a user updated from a v2 to a v3 registry,
+	// then pushed data into the registry again. in this situation, users will have
+	// data both, under /docker and under docker (without a leading /).
+	// in this case we assume the user has recovered from the apparent data loss,
+	// and wants to keep using the docker path (without a leading /).
+	// this means useLeadingSlash should be false.
+	contents = randStringRunes(4 * 1024 * 1024)
+	path2 := "/docker/registry/v2/repositories/test/hello/_layers/sha256/abc123/link"
+
+	defer d.Delete(ctx, path2)
+	// we have to manually set useLeadingSlash to false so that the path ends up
+	// not containing the leading slash - the driver framework doesn't allow us to
+	// create blobs that do not start with /
+	drvr.useLeadingSlash = false
+	if err := writeBlob(ctx, d, contents, path2, false); err != nil {
+		t.Fatal(err)
+	}
+
+	// create another instance of the storage driver. this is the only way
+	// to reset the value of useLeadingSlash
+	d, err = azureDriverConstructor()
+	if err != nil {
+		t.Fatalf("unexpected error creating azure driver: %v", err)
+	}
+	storageDriver = d.(*Driver)
+	drvr = storageDriver.Base.StorageDriver.(*driver)
+
+	if drvr.useLeadingSlash {
+		t.Error("expected a existing install of the registry with data under `docker` (without a leading slash) and `/docker` (with a leading slash) to default to use `docker` (without leading slash) but it default to `/docker`")
+	}
+	fmt.Println("========================================================================")
+}
+
+func TestListBlobs(t *testing.T) {
+	driver, err := azureDriverConstructor()
+	if err != nil {
+		t.Fatalf("unexpected error creating azure driver: %v", err)
+	}
+
+	contents := randStringRunes(4 * 1024 * 1024)
+	path := "/file"
+	ctx := context.Background()
+
+	defer driver.Delete(ctx, path)
+	if err := writeBlob(ctx, driver, contents, path, false); err != nil {
+		t.Fatal(err)
+	}
+
+	blobNames, err := driver.List(ctx, "/")
+	if err != nil {
+		t.Fatalf("driver.List: unexpected error: %v", err)
+	}
+
+	if len(blobNames) == 0 {
+		t.Fatalf("blob %q was not listed! driver.List returned 0 blobs!", path)
+	}
+
+	found := false
+	for _, blobName := range blobNames {
+		if blobName == path {
+			found = true
+			break
+		}
+		fmt.Println("TestListBlobs blobName", blobName, "did not match path", path)
+	}
+	if !found {
+		t.Fatalf("blob %q was not listed!", path)
+	}
+}
+
 func TestCommitAfterMove(t *testing.T) {
 	driver, err := azureDriverConstructor()
 	if err != nil {