import org.elasticsearch.common.blobstore.BlobPath; //导入依赖的package包/类
/**
 * Constructs a read-only URL-based repository
 */
public URLRepository(RepositoryMetaData metadata, Environment environment,
                     NamedXContentRegistry namedXContentRegistry) throws IOException {
    super(metadata, environment.settings(), namedXContentRegistry);

    if (URL_SETTING.exists(metadata.settings()) == false && REPOSITORIES_URL_SETTING.exists(settings) ==  false) {
        throw new RepositoryException(metadata.name(), "missing url");
    }
    supportedProtocols = SUPPORTED_PROTOCOLS_SETTING.get(settings);
    urlWhiteList = ALLOWED_URLS_SETTING.get(settings).toArray(new URIPattern[]{});
    this.environment = environment;

    URL url = URL_SETTING.exists(metadata.settings()) ? URL_SETTING.get(metadata.settings()) : REPOSITORIES_URL_SETTING.get(settings);
    URL normalizedURL = checkURL(url);
    blobStore = new URLBlobStore(settings, normalizedURL);
    basePath = BlobPath.cleanPath();
}
 

import org.elasticsearch.common.blobstore.BlobPath; //导入依赖的package包/类
public void testContainerCreationAndDeletion() throws IOException {
    try(BlobStore store = newBlobStore()) {
        final BlobContainer containerFoo = store.blobContainer(new BlobPath().add("foo"));
        final BlobContainer containerBar = store.blobContainer(new BlobPath().add("bar"));
        byte[] data1 = randomBytes(randomIntBetween(10, scaledRandomIntBetween(1024, 1 << 16)));
        byte[] data2 = randomBytes(randomIntBetween(10, scaledRandomIntBetween(1024, 1 << 16)));
        writeBlob(containerFoo, "test", new BytesArray(data1));
        writeBlob(containerBar, "test", new BytesArray(data2));

        assertArrayEquals(readBlobFully(containerFoo, "test", data1.length), data1);
        assertArrayEquals(readBlobFully(containerBar, "test", data2.length), data2);

        assertTrue(containerFoo.blobExists("test"));
        assertTrue(containerBar.blobExists("test"));
        store.delete(new BlobPath());
        assertFalse(containerFoo.blobExists("test"));
        assertFalse(containerBar.blobExists("test"));
    }
}
 

import org.elasticsearch.common.blobstore.BlobPath; //导入依赖的package包/类
public void testWriteRead() throws IOException {
    try(BlobStore store = newBlobStore()) {
        final BlobContainer container = store.blobContainer(new BlobPath());
        byte[] data = randomBytes(randomIntBetween(10, scaledRandomIntBetween(1024, 1 << 16)));
        writeBlob(container, "foobar", new BytesArray(data));
        try (InputStream stream = container.readBlob("foobar")) {
            BytesRefBuilder target = new BytesRefBuilder();
            while (target.length() < data.length) {
                byte[] buffer = new byte[scaledRandomIntBetween(1, data.length - target.length())];
                int offset = scaledRandomIntBetween(0, buffer.length - 1);
                int read = stream.read(buffer, offset, buffer.length - offset);
                target.append(new BytesRef(buffer, offset, read));
            }
            assertEquals(data.length, target.length());
            assertArrayEquals(data, Arrays.copyOfRange(target.bytes(), 0, target.length()));
        }
    }
}
 

import org.elasticsearch.common.blobstore.BlobPath; //导入依赖的package包/类
public void testBlobStoreOperations() throws IOException {
    BlobStore blobStore = createTestBlobStore();
    BlobContainer blobContainer = blobStore.blobContainer(BlobPath.cleanPath());
    ChecksumBlobStoreFormat<BlobObj> checksumJSON = new ChecksumBlobStoreFormat<>(BLOB_CODEC, "%s", BlobObj::fromXContent,
        xContentRegistry(), false, XContentType.JSON);
    ChecksumBlobStoreFormat<BlobObj> checksumSMILE = new ChecksumBlobStoreFormat<>(BLOB_CODEC, "%s", BlobObj::fromXContent,
        xContentRegistry(), false, XContentType.SMILE);
    ChecksumBlobStoreFormat<BlobObj> checksumSMILECompressed = new ChecksumBlobStoreFormat<>(BLOB_CODEC, "%s", BlobObj::fromXContent,
        xContentRegistry(), true, XContentType.SMILE);

    // Write blobs in different formats
    checksumJSON.write(new BlobObj("checksum json"), blobContainer, "check-json");
    checksumSMILE.write(new BlobObj("checksum smile"), blobContainer, "check-smile");
    checksumSMILECompressed.write(new BlobObj("checksum smile compressed"), blobContainer, "check-smile-comp");

    // Assert that all checksum blobs can be read by all formats
    assertEquals(checksumJSON.read(blobContainer, "check-json").getText(), "checksum json");
    assertEquals(checksumSMILE.read(blobContainer, "check-json").getText(), "checksum json");
    assertEquals(checksumJSON.read(blobContainer, "check-smile").getText(), "checksum smile");
    assertEquals(checksumSMILE.read(blobContainer, "check-smile").getText(), "checksum smile");
    assertEquals(checksumJSON.read(blobContainer, "check-smile-comp").getText(), "checksum smile compressed");
    assertEquals(checksumSMILE.read(blobContainer, "check-smile-comp").getText(), "checksum smile compressed");
}
 

import org.elasticsearch.common.blobstore.BlobPath; //导入依赖的package包/类
public void testCompressionIsApplied() throws IOException {
    BlobStore blobStore = createTestBlobStore();
    BlobContainer blobContainer = blobStore.blobContainer(BlobPath.cleanPath());
    StringBuilder veryRedundantText = new StringBuilder();
    for (int i = 0; i < randomIntBetween(100, 300); i++) {
        veryRedundantText.append("Blah ");
    }
    ChecksumBlobStoreFormat<BlobObj> checksumFormat = new ChecksumBlobStoreFormat<>(BLOB_CODEC, "%s", BlobObj::fromXContent,
        xContentRegistry(), false, randomBoolean() ? XContentType.SMILE : XContentType.JSON);
    ChecksumBlobStoreFormat<BlobObj> checksumFormatComp = new ChecksumBlobStoreFormat<>(BLOB_CODEC, "%s", BlobObj::fromXContent,
        xContentRegistry(), true, randomBoolean() ? XContentType.SMILE : XContentType.JSON);
    BlobObj blobObj = new BlobObj(veryRedundantText.toString());
    checksumFormatComp.write(blobObj, blobContainer, "blob-comp");
    checksumFormat.write(blobObj, blobContainer, "blob-not-comp");
    Map<String, BlobMetaData> blobs = blobContainer.listBlobsByPrefix("blob-");
    assertEquals(blobs.size(), 2);
    assertThat(blobs.get("blob-not-comp").length(), greaterThan(blobs.get("blob-comp").length()));
}
 

import org.elasticsearch.common.blobstore.BlobPath; //导入依赖的package包/类
/**
 * Constructs new shared file system repository
 *
 * @param name                 repository name
 * @param repositorySettings   repository settings
 * @param indexShardRepository index shard repository
 */
@Inject
public FsRepository(RepositoryName name, RepositorySettings repositorySettings, IndexShardRepository indexShardRepository, Environment environment) throws IOException {
    super(name.getName(), repositorySettings, indexShardRepository);
    Path locationFile;
    String location = repositorySettings.settings().get("location", settings.get("repositories.fs.location"));
    if (location == null) {
        logger.warn("the repository location is missing, it should point to a shared file system location that is available on all master and data nodes");
        throw new RepositoryException(name.name(), "missing location");
    }
    locationFile = environment.resolveRepoFile(location);
    if (locationFile == null) {
        if (environment.repoFiles().length > 0) {
            logger.warn("The specified location [{}] doesn't start with any repository paths specified by the path.repo setting: [{}] ", location, environment.repoFiles());
            throw new RepositoryException(name.name(), "location [" + location + "] doesn't match any of the locations specified by path.repo");
        } else {
            logger.warn("The specified location [{}] should start with a repository path specified by the path.repo setting, but the path.repo setting was not set on this node", location);
            throw new RepositoryException(name.name(), "location [" + location + "] doesn't match any of the locations specified by path.repo because this setting is empty");
        }
    }

    blobStore = new FsBlobStore(settings, locationFile);
    this.chunkSize = repositorySettings.settings().getAsBytesSize("chunk_size", settings.getAsBytesSize("repositories.fs.chunk_size", null));
    this.compress = repositorySettings.settings().getAsBoolean("compress", settings.getAsBoolean("repositories.fs.compress", false));
    this.basePath = BlobPath.cleanPath();
}
 

import org.elasticsearch.common.blobstore.BlobPath; //导入依赖的package包/类
/**
 * {@inheritDoc}
 */
@Override
public void initializeSnapshot(SnapshotId snapshotId, List<String> indices, MetaData metaData) {
    if (readOnly()) {
        throw new RepositoryException(this.repositoryName, "cannot create snapshot in a readonly repository");
    }
    try {
        if (snapshotFormat.exists(snapshotsBlobContainer, snapshotId.getSnapshot()) ||
                snapshotLegacyFormat.exists(snapshotsBlobContainer, snapshotId.getSnapshot())) {
            throw new InvalidSnapshotNameException(snapshotId, "snapshot with such name already exists");
        }
        // Write Global MetaData
        globalMetaDataFormat.write(metaData, snapshotsBlobContainer, snapshotId.getSnapshot());
        for (String index : indices) {
            final IndexMetaData indexMetaData = metaData.index(index);
            final BlobPath indexPath = basePath().add("indices").add(index);
            final BlobContainer indexMetaDataBlobContainer = blobStore().blobContainer(indexPath);
            indexMetaDataFormat.write(indexMetaData, indexMetaDataBlobContainer, snapshotId.getSnapshot());
        }
    } catch (IOException ex) {
        throw new SnapshotCreationException(snapshotId, ex);
    }
}
 

import org.elasticsearch.common.blobstore.BlobPath; //导入依赖的package包/类
@Test
public void testListBlobs() throws IOException {
    String containerName = "list_test_container";
    containers.add(containerName);

    CloudFilesBlobStore blobStore = getBlobStore(containerName);
    ImmutableBlobContainer immutableBlobContainer = blobStore.immutableBlobContainer(new BlobPath());

    // create test files
    String[] fileNames = new String[]{"test.txt", "foo.txt", "bar.txt", "bob.txt"};
    for(String fileName : fileNames){
        blobStoreContext.getBlobStore().putBlob(containerName, blobStoreContext.getBlobStore().blobBuilder(fileName)
                .payload(new ByteArrayInputStream("foobar".getBytes("UTF-8"))).build());
    }

    immutableBlobContainer.listBlobs();
}
 

import org.elasticsearch.common.blobstore.BlobPath; //导入依赖的package包/类
/**
 * Constructs new BlobStoreRepository
 * @param name                 repository name
 * @param repositorySettings   repository settings
 * @param indexShardRepository an instance of IndexShardRepository
 * @param gridFsService and instance of GridFsService
 */
@Inject
protected GridFsRepository(RepositoryName name, RepositorySettings repositorySettings, IndexShardRepository indexShardRepository, GridFsService gridFsService) {
    super(name.getName(), repositorySettings, indexShardRepository);

    String database = repositorySettings.settings().get("database", componentSettings.get("database"));
    if (database == null) {
        throw new RepositoryException(name.name(), "No database defined for GridFS repository");
    }

    String bucket = repositorySettings.settings().get("bucket", "fs");
    String host = repositorySettings.settings().get("gridfs_host", "localhost");
    int port = repositorySettings.settings().getAsInt("gridfs_port", 27017);
    String username = repositorySettings.settings().get("gridfs_username");
    String password = repositorySettings.settings().get("gridfs_password");

    int concurrentStreams = repositorySettings.settings().getAsInt("concurrent_streams", componentSettings.getAsInt("concurrent_streams", 5));
    ExecutorService concurrentStreamPool = EsExecutors.newScaling(1, concurrentStreams, 5, TimeUnit.SECONDS, EsExecutors.daemonThreadFactory(settings, "[gridfs_stream]"));
    blobStore = new GridFsBlobStore(settings, gridFsService.mongoDB(host, port, database, username, password), bucket, concurrentStreamPool);
    this.chunkSize = repositorySettings.settings().getAsBytesSize("chunk_size", componentSettings.getAsBytesSize("chunk_size", null));
    this.compress = repositorySettings.settings().getAsBoolean("compress", componentSettings.getAsBoolean("compress", true));
    this.basePath = BlobPath.cleanPath();
}
 

import org.elasticsearch.common.blobstore.BlobPath; //导入依赖的package包/类
/**
 * Delete an arbitrary BlobPath from our store.
 * @param path The blob path to delete
 */
@Override
public void delete(final BlobPath path) {
    SwiftPerms.exec(new PrivilegedAction<Void>() {
        @Override
        public Void run() {
            String keyPath = path.buildAsString("/");
            if (!keyPath.isEmpty()) {
                keyPath = keyPath + "/";
            }
            StoredObject obj = swift().getObject(keyPath);
            if (obj.exists()) {
                obj.delete();
            }
            return null;
        }
    });
}
 

import org.elasticsearch.common.blobstore.BlobPath; //导入依赖的package包/类
public InputStream get(final BlobPath blobPath) throws SftpException, JSchException {
    final Session session = sshPool.getSession();
    final ChannelSftp channel = openSftpChannel(session);
    final InputStream is = channel.get(config.getLocation() + "/"
        + blobPath.buildAsString("/"));
    return new InputStream() {
        @Override
        public int read() throws IOException {
            return is.read();
        }

        @Override
        public int read(byte b[], int off, int len) throws IOException {
            return is.read(b, off, len);
        }

        @Override
        public void close() throws IOException {
            is.close();
            closeChannel(channel);
            sshPool.returnSession(session);
        }

    };
}
 

import org.elasticsearch.common.blobstore.BlobPath; //导入依赖的package包/类
@Inject
public SshRepository(final RepositoryName name,
    final RepositorySettings repositorySettings,
    final IndexShardRepository indexShardRepository,
    final ThreadPool threadPool) throws IOException {
    super(name.getName(), repositorySettings, indexShardRepository);

    try {
        blobStore = new SshBlobStore(settings, new JSchClient(
            settings, repositorySettings, threadPool));
    } catch (final JSchException e) {
        throw new RepositoryException(name.name(),
            "Failed to initialize SSH configuration.", e);
    }

    chunkSize = repositorySettings.settings().getAsBytesSize("chunk_size",
        settings.getAsBytesSize("chunk_size", null));
    compress = repositorySettings.settings().getAsBoolean("compress",
        settings.getAsBoolean("compress", false));
    basePath = BlobPath.cleanPath();
}
 

import org.elasticsearch.common.blobstore.BlobPath; //导入依赖的package包/类
/**
 * {@inheritDoc}
 */
@Override
public BlobContainer blobContainer(BlobPath path) {
    try {
        return new URLBlobContainer(this, path, buildPath(path));
    } catch (MalformedURLException ex) {
        throw new BlobStoreException("malformed URL " + path, ex);
    }
}
 

import org.elasticsearch.common.blobstore.BlobPath; //导入依赖的package包/类
public void testURLBlobStoreCanReadBlob() throws IOException {
    BlobContainer container = urlBlobStore.blobContainer(BlobPath.cleanPath().add("indices"));
    try (InputStream stream = container.readBlob(blobName)) {
        byte[] bytes = new byte[message.length];
        int read = stream.read(bytes);
        assertEquals(message.length, read);
        assertArrayEquals(message, bytes);
    }
}
 

import org.elasticsearch.common.blobstore.BlobPath; //导入依赖的package包/类
public void testNoBlobFound() throws IOException {
    BlobContainer container = urlBlobStore.blobContainer(BlobPath.cleanPath().add("indices"));
    String incorrectBlobName = "incorrect_" + blobName;
    try (InputStream ignored = container.readBlob(incorrectBlobName)) {
        fail("Should have thrown NoSuchFileException exception");
        ignored.read();
    } catch (NoSuchFileException e) {
        assertEquals(String.format("[%s] blob not found", incorrectBlobName), e.getMessage());
    }
}
 

import org.elasticsearch.common.blobstore.BlobPath; //导入依赖的package包/类
public void testDeleteBlob() throws IOException {
    try (BlobStore store = newBlobStore()) {
        final String blobName = "foobar";
        final BlobContainer container = store.blobContainer(new BlobPath());
        expectThrows(IOException.class, () -> container.deleteBlob(blobName));

        byte[] data = randomBytes(randomIntBetween(10, scaledRandomIntBetween(1024, 1 << 16)));
        final BytesArray bytesArray = new BytesArray(data);
        writeBlob(container, blobName, bytesArray);
        container.deleteBlob(blobName); // should not raise

        // blob deleted, so should raise again
        expectThrows(IOException.class, () -> container.deleteBlob(blobName));
    }
}
 

import org.elasticsearch.common.blobstore.BlobPath; //导入依赖的package包/类
public void testVerifyOverwriteFails() throws IOException {
    try (BlobStore store = newBlobStore()) {
        final String blobName = "foobar";
        final BlobContainer container = store.blobContainer(new BlobPath());
        byte[] data = randomBytes(randomIntBetween(10, scaledRandomIntBetween(1024, 1 << 16)));
        final BytesArray bytesArray = new BytesArray(data);
        writeBlob(container, blobName, bytesArray);
        // should not be able to overwrite existing blob
        expectThrows(IOException.class, () -> writeBlob(container, blobName, bytesArray));
        container.deleteBlob(blobName);
        writeBlob(container, blobName, bytesArray); // after deleting the previous blob, we should be able to write to it again
    }
}
 

import org.elasticsearch.common.blobstore.BlobPath; //导入依赖的package包/类
/**
 * Constructs a shared file system repository.
 */
public FsRepository(RepositoryMetaData metadata, Environment environment,
                    NamedXContentRegistry namedXContentRegistry) throws IOException {
    super(metadata, environment.settings(), namedXContentRegistry);
    String location = REPOSITORIES_LOCATION_SETTING.get(metadata.settings());
    if (location.isEmpty()) {
        logger.warn("the repository location is missing, it should point to a shared file system location that is available on all master and data nodes");
        throw new RepositoryException(metadata.name(), "missing location");
    }
    Path locationFile = environment.resolveRepoFile(location);
    if (locationFile == null) {
        if (environment.repoFiles().length > 0) {
            logger.warn("The specified location [{}] doesn't start with any repository paths specified by the path.repo setting: [{}] ", location, environment.repoFiles());
            throw new RepositoryException(metadata.name(), "location [" + location + "] doesn't match any of the locations specified by path.repo");
        } else {
            logger.warn("The specified location [{}] should start with a repository path specified by the path.repo setting, but the path.repo setting was not set on this node", location);
            throw new RepositoryException(metadata.name(), "location [" + location + "] doesn't match any of the locations specified by path.repo because this setting is empty");
        }
    }

    blobStore = new FsBlobStore(settings, locationFile);
    if (CHUNK_SIZE_SETTING.exists(metadata.settings())) {
        this.chunkSize = CHUNK_SIZE_SETTING.get(metadata.settings());
    } else if (REPOSITORIES_CHUNK_SIZE_SETTING.exists(settings)) {
        this.chunkSize = REPOSITORIES_CHUNK_SIZE_SETTING.get(settings);
    } else {
        this.chunkSize = null;
    }
    this.compress = COMPRESS_SETTING.exists(metadata.settings()) ? COMPRESS_SETTING.get(metadata.settings()) : REPOSITORIES_COMPRESS_SETTING.get(settings);
    this.basePath = BlobPath.cleanPath();
}
 

import org.elasticsearch.common.blobstore.BlobPath; //导入依赖的package包/类
@Override
public void initializeSnapshot(SnapshotId snapshotId, List<IndexId> indices, MetaData clusterMetaData) {
    if (isReadOnly()) {
        throw new RepositoryException(metadata.name(), "cannot create snapshot in a readonly repository");
    }
    try {
        final String snapshotName = snapshotId.getName();
        // check if the snapshot name already exists in the repository
        final RepositoryData repositoryData = getRepositoryData();
        if (repositoryData.getAllSnapshotIds().stream().anyMatch(s -> s.getName().equals(snapshotName))) {
            throw new InvalidSnapshotNameException(metadata.name(), snapshotId.getName(), "snapshot with the same name already exists");
        }
        if (snapshotFormat.exists(snapshotsBlobContainer, snapshotId.getUUID())) {
            throw new InvalidSnapshotNameException(metadata.name(), snapshotId.getName(), "snapshot with the same name already exists");
        }

        // Write Global MetaData
        globalMetaDataFormat.write(clusterMetaData, snapshotsBlobContainer, snapshotId.getUUID());

        // write the index metadata for each index in the snapshot
        for (IndexId index : indices) {
            final IndexMetaData indexMetaData = clusterMetaData.index(index.getName());
            final BlobPath indexPath = basePath().add("indices").add(index.getId());
            final BlobContainer indexMetaDataBlobContainer = blobStore().blobContainer(indexPath);
            indexMetaDataFormat.write(indexMetaData, indexMetaDataBlobContainer, snapshotId.getUUID());
        }
    } catch (IOException ex) {
        throw new SnapshotCreationException(metadata.name(), snapshotId, ex);
    }
}
 

import org.elasticsearch.common.blobstore.BlobPath; //导入依赖的package包/类
@Override
public BlobContainer blobContainer(BlobPath path) {
    try {
        return new FsBlobContainer(this, path, buildAndCreate(path));
    } catch (IOException ex) {
        throw new ElasticsearchException("failed to create blob container", ex);
    }
}