java.nio.file.AtomicMoveNotSupportedException#org.sonatype.nexus.blobstore.api.Blob源码实例Demo

private Blob createMockBlob(String id) {
  String createdBy = "createdBy";
  String createdByIp = "createdByIp";
  DateTime creationTime = new DateTime();
  Blob blob = mock(Blob.class);

  Map<String, String> headers = new HashMap<>();
  headers.put(CREATED_BY_HEADER, createdBy);
  headers.put(CREATED_BY_IP_HEADER, createdByIp);
  when(blob.getHeaders()).thenReturn(headers);

  BlobMetrics metrics = new BlobMetrics(creationTime, "hash", 1L);
  when(blob.getMetrics()).thenReturn(metrics);

  BlobId blobId = new BlobId(id);
  when(blob.getId()).thenReturn(blobId);
  when(blobStore.get(blobId)).thenReturn(blob);

  return blob;
}
 

@Before
public void setUp() {
  Blob restoredBlob = mockBlob(RESTORED_BLOB_ID);
  Blob copiedBlob = mockBlob(COPIED_BLOB_ID);

  when(blobStore.create(blobData, headers, null)).thenAnswer(this::newBlob);
  when(blobStore.create(blobData, headers, RESTORED_BLOB_ID)).thenReturn(restoredBlob);
  when(blobStore.create(sourceFile, headers, TEST_BLOB_SIZE, TEST_BLOB_HASH)).thenAnswer(this::newBlob);
  when(blobStore.copy(EXISTING_BLOB_ID, headers)).thenReturn(copiedBlob);

  when(blobStore.exists(any())).thenReturn(true);
  when(blobStore.get(any())).thenAnswer(this::getBlob);

  BlobStoreConfiguration storeConfiguration = mock(BlobStoreConfiguration.class);
  when(storeConfiguration.getName()).thenReturn("test-blob-store");
  when(blobStore.getBlobStoreConfiguration()).thenReturn(storeConfiguration);
}
 

/**
 * Extracts the contents for the first matching {@code composer.json} file (of which there should only be one) as a
 * map representing the parsed JSON content. If no such file is found then an empty map is returned.
 */
public Map<String, Object> extractFromZip(final Blob blob) throws IOException {
  try (InputStream is = blob.getInputStream()) {
    try (ArchiveInputStream ais = archiveStreamFactory.createArchiveInputStream(ArchiveStreamFactory.ZIP, is)) {
      ArchiveEntry entry = ais.getNextEntry();
      while (entry != null) {
        Map<String, Object> contents = processEntry(ais, entry);
        if (!contents.isEmpty()) {
          return contents;
        }
        entry = ais.getNextEntry();
      }
    }
    return Collections.emptyMap();
  }
  catch (ArchiveException e) {
    throw new IOException("Error reading from archive", e);
  }
}
 

private PrefetchedAssetBlob createPrefetchedAssetBlob(final Blob blob,
                                                      final Map<HashAlgorithm, HashCode> hashes,
                                                      final boolean hashesVerified,
                                                      final String contentType)
{
  PrefetchedAssetBlob assetBlob = new PrefetchedAssetBlob(
      nodeAccess,
      blobStore,
      blob,
      contentType,
      hashes,
      hashesVerified);

  newlyCreatedBlobs.add(assetBlob);
  return assetBlob;
}
 

@Test
public void hardDeletePreventsGetDespiteOpenStreams() throws Exception {
  final byte[] content = new byte[TEST_DATA_LENGTH];
  new Random().nextBytes(content);

  final Blob blob = underTest.create(new ByteArrayInputStream(content), TEST_HEADERS);

  final InputStream inputStream = blob.getInputStream();

  // Read half the data
  inputStream.read(new byte[content.length / 2]);

  // force delete
  underTest.deleteHard(blob.getId());

  final Blob newBlob = underTest.get(blob.getId());
  assertThat(newBlob, is(nullValue()));
}
 

@Override
public TempBlob ingest(final InputStream in,
                       @Nullable final String contentType,
                       final Iterable<HashAlgorithm> hashing)
{
  Optional<ClientInfo> clientInfo = facet.clientInfo();

  Builder<String, String> tempHeaders = ImmutableMap.builder();
  tempHeaders.put(TEMPORARY_BLOB_HEADER, "");
  tempHeaders.put(REPO_NAME_HEADER, facet.repository().getName());
  tempHeaders.put(BLOB_NAME_HEADER, "temp");
  tempHeaders.put(CREATED_BY_HEADER, clientInfo.map(ClientInfo::getUserid).orElse("system"));
  tempHeaders.put(CREATED_BY_IP_HEADER, clientInfo.map(ClientInfo::getRemoteIP).orElse("system"));
  tempHeaders.put(CONTENT_TYPE_HEADER, ofNullable(contentType).orElse(APPLICATION_OCTET_STREAM));

  MultiHashingInputStream hashingStream = new MultiHashingInputStream(hashing, in);
  Blob blob = facet.stores().blobStore.create(hashingStream, tempHeaders.build());

  return new TempBlob(blob, hashingStream.hashes(), true, facet.stores().blobStore);
}
 

private void runBlobRestore(final boolean isDryRun) {
  Asset asset;
  Blob blob;
  try (StorageTx tx = getStorageTx(proxyRepository)) {
    tx.begin();
    asset = tx.findAssetWithProperty(AssetEntityAdapter.P_NAME, MONGO_PATH_FULL_728_TARGZ,
        tx.findBucket(proxyRepository));
    assertThat(asset, Matchers.notNullValue());
    blob = tx.getBlob(asset.blobRef());
  }
  testHelper.simulateAssetMetadataLoss();
  Properties properties = new Properties();
  properties.setProperty(HEADER_PREFIX + REPO_NAME_HEADER, proxyRepository.getName());
  properties.setProperty(HEADER_PREFIX + BLOB_NAME_HEADER, asset.name());
  properties.setProperty(HEADER_PREFIX + CONTENT_TYPE_HEADER, asset.contentType());

  helmRestoreBlobStrategy.restore(properties, blob, BlobStoreManager.DEFAULT_BLOBSTORE_NAME, isDryRun);
}
 

private void runBlobRestore(final boolean isDryRun) {
  Asset asset;
  Blob blob;
  try (StorageTx tx = getStorageTx(proxyRepository)) {
    tx.begin();
    asset = tx.findAssetWithProperty(AssetEntityAdapter.P_NAME, AGRICOLAE_131_TARGZ.fullPath,
        tx.findBucket(proxyRepository));
    assertThat(asset, Matchers.notNullValue());
    blob = tx.getBlob(asset.blobRef());
  }
  testHelper.simulateAssetMetadataLoss();
  Properties properties = new Properties();
  properties.setProperty(HEADER_PREFIX + REPO_NAME_HEADER, proxyRepository.getName());
  properties.setProperty(HEADER_PREFIX + BLOB_NAME_HEADER, asset.name());
  properties.setProperty(HEADER_PREFIX + CONTENT_TYPE_HEADER, asset.contentType());

  rRestoreBlobStrategy.restore(properties, blob, BlobStoreManager.DEFAULT_BLOBSTORE_NAME, isDryRun);
}
 

@Override
@Guarded(by = STARTED)
public Blob create(final InputStream blobData, final Map<String, String> headers) {
  checkNotNull(blobData);

  return create(headers, destination -> {
      try (InputStream data = blobData) {
        MetricsInputStream input = new MetricsInputStream(data);
        TransferManager transferManager = new TransferManager(s3);
        transferManager.upload(getConfiguredBucket(), destination, input, new ObjectMetadata())
            .waitForCompletion();
        return input.getMetrics();
      } catch (InterruptedException e) {
        throw new BlobStoreException("error uploading blob", e, null);
      }
    });
}
 

@Nullable
@Override
@Guarded(by = STARTED)
public Blob get(final BlobId blobId, final boolean includeDeleted) {
  if (includeDeleted) {
    // check directly without using cache
    return members.get().stream()
        .map((BlobStore member) -> member.get(blobId, true))
        .filter(Objects::nonNull)
        .findAny()
        .orElse(null);
  }
  else {
    return locate(blobId)
      .map((BlobStore target) -> target.get(blobId, false))
      .orElse(null);
  }
}
 

@Override
public void assertAssetMatchesBlob(final Repository repository, final String... names) {
  for (String name : names) {
    try (StorageTx tx = getStorageTx(repository)) {
      tx.begin();
      Asset asset = tx.findAssetWithProperty(AssetEntityAdapter.P_NAME, name, tx.findBucket(repository));
      Blob blob = tx.requireBlob(asset.blobRef());

      assertThat(repository.getName(), equalTo(blob.getHeaders().get(Bucket.REPO_NAME_HEADER)));
      assertThat(asset.name(), equalTo(blob.getHeaders().get(BlobStore.BLOB_NAME_HEADER)));
      assertThat(asset.createdBy(), equalTo(blob.getHeaders().get(BlobStore.CREATED_BY_HEADER)));
      assertThat(asset.createdByIp(), equalTo(blob.getHeaders().get(BlobStore.CREATED_BY_IP_HEADER)));
      assertThat(asset.contentType(), equalTo(blob.getHeaders().get(BlobStore.CONTENT_TYPE_HEADER)));
      assertThat(asset.attributes().child("checksum").get("sha1"), equalTo(blob.getMetrics().getSha1Hash()));
      assertThat(asset.size(), equalTo(blob.getMetrics().getContentSize()));
    }
  }
}
 

private void updatePackageRootIfShaIncorrect(final Repository repository,
                                             final Asset asset,
                                             final Blob blob,
                                             final NestedAttributesMap newPackageRoot,
                                             final NpmPackageId packageId,
                                             final String packageVersion)
{
  NpmHostedFacet hostedFacet = repository.facet(NpmHostedFacet.class);
  try {
    NestedAttributesMap oldPackageRoot = getPackageRoot(UnitOfWork.currentTx(), repository, packageId);
    if (oldPackageRoot != null) {
      String oldSha = extractShasum(oldPackageRoot, packageVersion);
      String newSha = extractShasum(newPackageRoot, packageVersion);

      if (!Objects.equals(oldSha, newSha)) {
        maybeUpdateIntegrity(asset, blob, packageVersion, oldPackageRoot, newPackageRoot);

        hostedFacet.putPackageRoot(packageId, null, newPackageRoot);
      }
    }
  }
  catch (IOException e) {
    log.error("Failed to update asset {}", asset.name(), e);
  }
}
 

private String calculateIntegrity(final Asset asset, final Blob blob, final String algorithm) {
  try {
    HashCode hash;
    if (algorithm.equalsIgnoreCase(SHA1.name())) {
      hash = hash(SHA1, blob.getInputStream());
    }
    else {
      hash = hash(SHA512, blob.getInputStream());
    }

    return algorithm + "-" + Base64.getEncoder().encodeToString(hash.asBytes());
  }
  catch (IOException e) {
    log.error("Failed to calculate hash for asset {}", asset.name(), e);
  }
  return "";
}
 

@Test
public void getDirectPathBlobIdStreamSuccess() throws IOException {
  byte[] content = "hello".getBytes();
  Blob blob = underTest.create(new ByteArrayInputStream(content), ImmutableMap.of(
      CREATED_BY_HEADER, "test",
      BLOB_NAME_HEADER, "health-check/repositoryName/file.txt",
      DIRECT_PATH_BLOB_HEADER, "true"
  ));
  verifyMoveOperationsAtomic(blob);

  assertThat(underTest.getDirectPathBlobIdStream("health-check").count(), is(1L));
  // confirm same result for deeper prefix
  assertThat(underTest.getDirectPathBlobIdStream("health-check/repositoryName").count(), is(1L));
  // confirm same result for the top
  assertThat(underTest.getDirectPathBlobIdStream(".").count(), is(1L));
  assertThat(underTest.getDirectPathBlobIdStream("").count(), is(1L));

  BlobId blobId = underTest.getDirectPathBlobIdStream("health-check").findFirst().get();
  assertThat(blobId, is(blob.getId()));

  // this check is more salient when run on Windows but confirms that direct path BlobIds use unix style paths
  assertThat(blobId.asUniqueString().contains("\\"), is(false));
  assertThat(blobId.asUniqueString().contains("/"), is(true));
}
 

/**
 * Verify that the Blob exists and is in agreement with the stored Asset metadata.;
 */
private void verifyBlob(final Blob blob, final Asset asset) throws MismatchedSHA1Exception, BlobUnavilableException, IOException {
  BlobMetrics metrics = blob.getMetrics();

  String assetChecksum =
      asset.blob().map(AssetBlob::checksums).map(checksums -> checksums.get(HashAlgorithm.SHA1.name())).orElse(null);
  if (!metrics.getSha1Hash().equals(assetChecksum)) {
    throw new MismatchedSHA1Exception();
  }

  try (InputStream blobstream = blob.getInputStream()) {
    if (metrics.getContentSize() > 0 && blobstream.available() == 0) {
      throw new BlobUnavilableException();
    }
  }
}
 

private void mockPackageMetadata() {
  Asset packageAssetRoot = mock(Asset.class);
  when(packageAssetRoot.name()).thenReturn("@foo/bar");
  when(packageAssetRoot.getEntityMetadata())
      .thenReturn(new DetachedEntityMetadata(new DetachedEntityId("foo"), new DetachedEntityVersion("a")));
  when(packageAssetRoot.formatAttributes()).thenReturn(new NestedAttributesMap("metadata", new HashMap<>()));
  BlobRef blobRef = mock(BlobRef.class);
  when(packageAssetRoot.requireBlobRef()).thenReturn(blobRef);
  Blob blob = mock(Blob.class);
  when(storageTx.requireBlob(blobRef)).thenReturn(blob);

  when(blob.getInputStream())
      .thenReturn(new ByteArrayInputStream("{\"name\": \"@foo/bar\",\"versions\": {\"0.1\": {}}}".getBytes()));

  when(storageTx.findAssetWithProperty(eq(P_NAME), eq("@foo/bar"), any(Bucket.class))).thenReturn(packageAssetRoot);

}
 

private AssetBlob createAssetBlob(final Function<BlobStore, Blob> blobFunction,
                                  final Map<HashAlgorithm, HashCode> hashes,
                                  final boolean hashesVerified,
                                  final String contentType)
{
  AssetBlob assetBlob = new AssetBlob(
      nodeAccess,
      blobStore,
      blobFunction,
      contentType,
      hashes,
      hashesVerified);

  newlyCreatedBlobs.add(assetBlob);
  return assetBlob;
}
 

private void rebuildMetadata(final Blob metadataBlob) throws IOException {
  Metadata metadata = MavenModels.readMetadata(metadataBlob.getInputStream());

  // avoid triggering nested rebuilds as the rebuilder will already do that if necessary
  rebuilding.set(TRUE);
  try {
    metadataRebuilder.rebuildInTransaction(
        getRepository(),
        false,
        false,
        metadata.getGroupId(),
        metadata.getArtifactId(),
        metadata.getVersion()
    );
  }
  finally {
    rebuilding.remove();
  }
}
 

@Test
public void testGet_expiredMetadata() throws Exception {
  String path = "/org/sonatype/nexus/nexus-base/3.19.0-SNAPSHOT/maven-metadata.xml";
  Asset asset = createMetadataAsset(path, INVALIDATED);
  Blob blob = mock(Blob.class);
  when(blob.getInputStream())
      .thenReturn(getClass().getResourceAsStream("/org/sonatype/nexus/repository/maven/gavMetadata.xml"));
  when(storageTx.findAssetWithProperty(any(), any(), any(Bucket.class))).thenReturn(asset);
  when(storageTx.requireBlob(any())).thenReturn(blob);
  MavenPath mavenPath = maven2MavenPathParser.parsePath(path);
  Content content = underTest.get(mavenPath);

  assertThat(content, not(nullValue()));
  assertThat(content.getContentType(), is(TEXT_XML));
  verify(metadataRebuilder)
      .rebuildInTransaction(any(), eq(false), eq(false), eq("org.sonatype.nexus"), eq("nexus-base"),
          eq("3.19.0-SNAPSHOT"));
}
 

@Test
public void testGet_currentMetadata() throws Exception {
  String path = "/org/sonatype/nexus/nexus-base/3.19.0-SNAPSHOT/maven-metadata.xml";
  Asset asset = createMetadataAsset(path, "valid");
  Blob blob = mock(Blob.class);
  when(blob.getInputStream())
      .thenReturn(getClass().getResourceAsStream("/org/sonatype/nexus/repository/maven/gavMetadata.xml"));
  when(storageTx.findAssetWithProperty(any(), any(), any(Bucket.class))).thenReturn(asset);
  when(storageTx.requireBlob(any())).thenReturn(blob);
  MavenPath mavenPath = maven2MavenPathParser.parsePath(path);
  Content content = underTest.get(mavenPath);

  assertThat(content, not(nullValue()));
  assertThat(content.getContentType(), is(TEXT_XML));
  verify(storageTx, never()).deleteAsset(any());
  verify(metadataRebuilder, never())
      .rebuildInTransaction(any(), eq(false), eq(false), eq("org.sonatype.nexus"), eq("nexus-base"),
          eq("3.19.0-SNAPSHOT"));
}
 

@Test
public void anAssetBlobCanBeDeletedWhileTheSystemIsInspected() {
  AssetBlob missingAssetBlob = mockAssetBlob(mock(AssetBlob.class));
  when(asset.blob()).thenReturn(Optional.of(missingAssetBlob)); // first pass we have a missing blobRef

  FluentAsset reloadedAsset = createAsset(assetBlob);
  Blob reloadedBlob = mock(Blob.class); // second pass the blobRef is there but file does not exist
  when(reloadedBlob.getMetrics()).thenReturn(blobMetrics);
  BlobId missingBlobId = reloadedAsset.blob().get().blobRef().getBlobId();
  when(blobStore.get(missingBlobId)).thenReturn(reloadedBlob);

  mockAssetBrowse();
  mockAssetReload(reloadedAsset);

  when(reloadedBlob.getMetrics()).thenReturn(blobMetrics);
  when(reloadedBlob.getInputStream()).thenThrow(new BlobStoreException("Blob has been deleted", new BlobId("foo")));

  List<DeadBlobResult<Asset>> result = deadBlobFinder.find(repository, true);

  assertThat(result, hasSize(1));
  assertThat(result.get(0).getResultState(), is(DELETED));
}
 

@Override
public void restore(final Properties properties,
                    final Blob blob,
                    final String blobStoreName,
                    final boolean isDryRun)
{
  RestoreBlobData blobData = new RestoreBlobData(blob, properties, blobStoreName, repositoryManager);
  Optional<StorageFacet> storageFacet = blobData.getRepository().optionalFacet(StorageFacet.class);
  T restoreData = createRestoreData(blobData);

  if (storageFacet.isPresent() && canAttemptRestore(restoreData)) {
    doRestore(storageFacet.get(), blobData, restoreData, isDryRun);
  }
  else {
    log.debug("Skipping asset, blob store: {}, repository: {}, blob name: {}, blob id: {}",
        blobStoreName, blobData.getRepository().getName(), blobData.getBlobName(), blob.getId());
  }
}
 

@Override
protected Blob doCreate(final InputStream blobData,
                        final Map<String, String> headers,
                        @Nullable final BlobId blobId)
{
  return createInternal(headers, destination -> {
    try (InputStream data = blobData) {
      MetricsInputStream input = new MetricsInputStream(data);

      uploader.upload(storage, getConfiguredBucketName(), destination, input);
      return input.getMetrics();
    }
  }, blobId);
}
 

@Override
@Guarded(by = STARTED)
public Blob copy(final BlobId blobId, final Map<String, String> headers) {
  BlobStore target = locate(blobId)
      .orElseThrow(() -> new BlobStoreException("Unable to find blob", blobId));
  Blob blob = target.copy(blobId, headers);
  locatedBlobs.put(blob.getId(), target.getBlobStoreConfiguration().getName());
  return blob;
}
 

@Nullable
@Override
@Guarded(by = STARTED)
@Timed
public Blob get(final BlobId blobId, final boolean includeDeleted) {
  checkNotNull(blobId);

  final GoogleCloudStorageBlob blob = liveBlobs.getUnchecked(blobId);

  if (blob.isStale()) {
    Lock lock = blob.lock();
    try {
      if (blob.isStale()) {
        GoogleCloudBlobAttributes blobAttributes = new GoogleCloudBlobAttributes(bucket, attributePath(blobId));
        boolean loaded = blobAttributes.load();
        if (!loaded) {
          log.warn("Attempt to access non-existent blob {} ({})", blobId, blobAttributes);
          return null;
        }

        if (blobAttributes.isDeleted() && !includeDeleted) {
          log.warn("Attempt to access soft-deleted blob {} ({})", blobId, blobAttributes);
          return null;
        }

        blob.refresh(blobAttributes.getHeaders(), blobAttributes.getMetrics());
      }
    }
    catch (IOException e) {
      throw new BlobStoreException(e, blobId);
    }
    finally {
      lock.unlock();
    }
  }

  log.debug("Accessing blob {}", blobId);
  return blob;
}
 

@Override
@Guarded(by = ACTIVE)
public Blob requireBlob(final BlobRef blobRef) {
  Blob blob = getBlob(blobRef);
  if (blob == null) {
    throw new MissingBlobException(blobRef);
  }
  return blob;
}
 

/**
 * Convert an asset blob to {@link Content}.
 *
 * @return content of asset blob
 */
@Override
public Content toContent(final Asset asset, final Blob blob) {
  Content content = new Content(new BlobPayload(blob, asset.requireContentType()));
  Content.extractFromAsset(asset, HASH_ALGORITHMS, content.getAttributes());
  return content;
}
 

@Test
public void executeSkipsAssetsWithANonEmptyCreatedBy() {
  Blob blob = createMockBlob("blobId");
  Asset asset = createAsset(blob);
  asset.createdBy("somebody");
  assetStore.save(asset);

  task.execute();

  Asset updatedAsset = assetStore.getById(id(asset));
  assertThat(updatedAsset.name(), is(asset.name()));
  assertThat(updatedAsset.createdBy(), is("somebody"));
  assertThat(updatedAsset.createdByIp(), is(nullValue()));
}
 

public AssetBlob(final NodeAccess nodeAccess,
                 final BlobStore blobStore,
                 final Function<BlobStore, Blob> blobFunction,
                 final String contentType,
                 final Map<HashAlgorithm, HashCode> hashes,
                 final boolean hashesVerified)
{
  this.nodeAccess = checkNotNull(nodeAccess);
  this.blobStore = checkNotNull(blobStore);
  this.blobFunction = checkNotNull(blobFunction);
  this.contentType = checkNotNull(contentType);
  this.hashes = checkNotNull(hashes);
  this.hashesVerified = hashesVerified;
}
 

private Blob create(final Map<String, String> headers, final BlobIngester ingester) {
  checkNotNull(headers);

  checkArgument(headers.containsKey(BLOB_NAME_HEADER), "Missing header: %s", BLOB_NAME_HEADER);
  checkArgument(headers.containsKey(CREATED_BY_HEADER), "Missing header: %s", CREATED_BY_HEADER);

  final BlobId blobId = blobIdLocationResolver.fromHeaders(headers);

  final String blobPath = contentPath(blobId);
  final String attributePath = attributePath(blobId);
  final S3Blob blob = liveBlobs.getUnchecked(blobId);

  Lock lock = blob.lock();
  try {
    log.debug("Writing blob {} to {}", blobId, blobPath);

    final StreamMetrics streamMetrics = ingester.ingestTo(blobPath);
    final BlobMetrics metrics = new BlobMetrics(new DateTime(), streamMetrics.getSha1(), streamMetrics.getSize());
    blob.refresh(headers, metrics);

    S3BlobAttributes blobAttributes = new S3BlobAttributes(s3, getConfiguredBucket(), attributePath, headers, metrics);

    blobAttributes.store();
    storeMetrics.recordAddition(blobAttributes.getMetrics().getContentSize());

    return blob;
  }
  catch (IOException e) {
    // Something went wrong, clean up the files we created
    deleteQuietly(attributePath);
    deleteQuietly(blobPath);
    throw new BlobStoreException(e, blobId);
  }
  finally {
    lock.unlock();
  }
}