test archive file overwrite

deps/oblib/src/lib/ob_errno.h

@@ -373,6 +373,7 @@ constexpr int OB_BACKUP_META_INDEX_NOT_EXIST = -9076;
 constexpr int OB_BACKUP_DEVICE_OUT_OF_SPACE = -9082;
 constexpr int OB_BACKUP_PWRITE_OFFSET_NOT_MATCH = -9083;
 constexpr int OB_BACKUP_PWRITE_CONTENT_NOT_MATCH = -9084;
+constexpr int OB_CLOUD_OBJECT_NOT_APPENDABLE = -9098;
 constexpr int OB_ERR_XML_PARSE = -9549;
 constexpr int OB_ERR_XSLT_PARSE = -9574;
 constexpr int OB_MAX_RAISE_APPLICATION_ERROR         = -20000;

deps/oblib/src/lib/restore/ob_storage_oss_base.cpp

@@ -1955,7 +1955,7 @@ int ObStorageOssAppendWriter::do_write(const char *buf, const int64_t size, cons
           ret = OB_OSS_ERROR;
           OB_LOG(ERROR, "oss type is null", K(ret));
         } else if (0 != strncmp(OSS_OBJECT_TYPE_APPENDABLE, object_type, strlen(OSS_OBJECT_TYPE_APPENDABLE))) {
-          ret = OB_OSS_ERROR;
+          ret = OB_CLOUD_OBJECT_NOT_APPENDABLE;
           OB_LOG(WARN, "oss object not match", K(ret), KCSTRING(object_type));
         } else {
           char *next_append_position = (char*)(apr_table_get(resp_headers, OSS_NEXT_APPEND_POSITION));
@@ -1968,7 +1968,7 @@ int ObStorageOssAppendWriter::do_write(const char *buf, const int64_t size, cons
         }
       }
 
-      if (is_pwrite && position != offset) {
+      if (OB_SUCC(ret) && is_pwrite && position != offset) {
         ret = OB_BACKUP_PWRITE_OFFSET_NOT_MATCH;
         OB_LOG(WARN, "position and offset do not match", K(ret), K(position), K(offset));
       }

src/logservice/archiveservice/ob_archive_io.cpp

@@ -11,7 +11,9 @@
  */
 
 #include "ob_archive_io.h"
+#include "lib/ob_errno.h"
 #include "lib/oblog/ob_log_module.h"
+#include "lib/utility/ob_macro_utils.h"
 #include "lib/utility/ob_tracepoint.h"            // EventTable
 #include "share/ob_device_manager.h"              // ObIODevice
 #include "share/backup/ob_backup_io_adapter.h"    // ObBackupIoAdapter
@@ -77,6 +79,13 @@ int ObArchiveIO::push_log(const ObString &uri,
   if (OB_SUCCESS != (tmp_ret = util.close_device_and_fd(device_handle, fd))) {
     ARCHIVE_LOG(WARN, "fail to close file and release device!", K(tmp_ret), K(uri), KP(storage_info));
   }
+
+  if (OB_CLOUD_OBJECT_NOT_APPENDABLE == ret) {
+    if (OB_FAIL(check_context_match_in_normal_file_(uri, storage_info, data, data_len, offset))) {
+      ARCHIVE_LOG(WARN, "check_context_match_ failed", K(uri));
+    }
+  }
+
   return ret;
 }
 
@@ -86,5 +95,40 @@ int ObArchiveIO::mkdir(const ObString &uri, const share::ObBackupStorageInfo *st
   return util.mkdir(uri, storage_info);
 }
 
+int ObArchiveIO::check_context_match_in_normal_file_(const ObString &uri,
+    const share::ObBackupStorageInfo *storage_info,
+    char *data,
+    const int64_t data_len,
+    const int64_t offset)
+{
+  int ret = OB_SUCCESS;
+  int64_t length = 0;
+  char *read_buffer = NULL;
+  int64_t read_size = 0;
+  ObBackupIoAdapter reader;
+  ObArenaAllocator allocator;
+
+  if (OB_FAIL(reader.get_file_length(uri, storage_info, length))) {
+    ARCHIVE_LOG(WARN, "get file_length failed", K(uri));
+  } else if (OB_UNLIKELY(length <= offset)) {
+    ret = OB_BACKUP_PWRITE_CONTENT_NOT_MATCH;
+    ARCHIVE_LOG(ERROR, "object length smaller than offset", K(uri), K(length), K(offset));
+  } else if (OB_UNLIKELY(length < offset + data_len)) {
+    ret = OB_BACKUP_PWRITE_CONTENT_NOT_MATCH;
+    ARCHIVE_LOG(ERROR, "object length smaller than write buffer total offset", K(uri), K(length), K(offset), K(data_len));
+  } else if (OB_ISNULL(read_buffer = static_cast<char*>(allocator.alloc(data_len)))) {
+    ret = OB_ALLOCATE_MEMORY_FAILED;
+    ARCHIVE_LOG(WARN, "allocate memory failed", K(uri), K(data_len));
+  } else if (OB_FAIL(reader.read_part_file(uri, storage_info, read_buffer, data_len, offset, read_size))) {
+    ARCHIVE_LOG(WARN, "pread failed", K(uri), K(read_buffer), K(data_len), K(offset));
+  } else if (read_size < data_len) {
+    ret = OB_BACKUP_PWRITE_CONTENT_NOT_MATCH;
+    ARCHIVE_LOG(ERROR, "read_size smaller than data_len",  K(uri), K(length), K(offset), K(data_len), K(read_size));
+  } else if (0 != MEMCMP(data, read_buffer, data_len)) {
+    ret = OB_BACKUP_PWRITE_CONTENT_NOT_MATCH;
+    ARCHIVE_LOG(ERROR, "data inconsistent", K(uri), K(length), K(offset), K(data_len), K(read_size));
+  }
+  return ret;
+}
 } // namespace archive
 } // namespace oceanbase

src/logservice/archiveservice/ob_archive_io.h

@@ -37,6 +37,13 @@ public:
 
   int mkdir(const ObString &uri,
       const share::ObBackupStorageInfo *storage_info);
+
+private:
+  int check_context_match_in_normal_file_(const ObString &uri,
+      const share::ObBackupStorageInfo *storage_info,
+      char *data,
+      const int64_t data_len,
+      const int64_t offset);
 };
 } // namespace archive
 } // namespace oceanbase

src/share/ob_errno.def

@@ -2018,7 +2018,7 @@ DEFINE_ERROR(OB_LS_RESTORE_FAILED, -9094, -1, "HY000", "Restore log stream faile
 DEFINE_ERROR(OB_NO_TABLET_NEED_BACKUP, -9095, -1, "HY000", "No tablet need backup");
 DEFINE_ERROR(OB_ERR_RESTORE_STANDBY_VERSION_LAG, -9096, -1, "HY000", "standby binary version is lower than primary data version, standby need upgrade");
 DEFINE_ERROR(OB_ERR_RESTORE_PRIMARY_TENANT_DROPPED, -9097, -1, "HY000", "primary tenant has been dropped");
-DEFINE_ERROR(OB_CLOUD_OBJECT_NOT_APPENDABLE, -9098, -1, "HY000", "normal object in object_storage(oss,cos,etc.) can't be appended content");
+DEFINE_ERROR_DEP(OB_CLOUD_OBJECT_NOT_APPENDABLE, -9098, -1, "HY000", "normal object in object_storage(oss,cos,etc.) can't be appended content");
 // end of backup and restore error codes 9001 ~ 9099
 
 ////////////////////////////////////////////////////////////////

src/share/ob_errno.h

@@ -1499,7 +1499,6 @@ constexpr int OB_LS_RESTORE_FAILED = -9094;
 constexpr int OB_NO_TABLET_NEED_BACKUP = -9095;
 constexpr int OB_ERR_RESTORE_STANDBY_VERSION_LAG = -9096;
 constexpr int OB_ERR_RESTORE_PRIMARY_TENANT_DROPPED = -9097;
-constexpr int OB_CLOUD_OBJECT_NOT_APPENDABLE = -9098;
 constexpr int OB_NO_SUCH_FILE_OR_DIRECTORY = -9100;
 constexpr int OB_FILE_OR_DIRECTORY_EXIST = -9101;
 constexpr int OB_FILE_OR_DIRECTORY_PERMISSION_DENIED = -9102;

unittest/archive/CMakeLists.txt

@@ -1 +1,2 @@
 # ob_unittest(test_archive_task)
+ob_unittest(test_archive_overwrite)

unittest/archive/test_archive_overwrite.cpp

@@ -0,0 +1,87 @@
+/**
+ * Copyright (c) 2021 OceanBase
+ * OceanBase CE is licensed under Mulan PubL v2.
+ * You can use this software according to the terms and conditions of the Mulan PubL v2.
+ * You may obtain a copy of Mulan PubL v2 at:
+ *          http://license.coscl.org.cn/MulanPubL-2.0
+ * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
+ * EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
+ * MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
+ * See the Mulan PubL v2 for more details.
+ */
+
+#include "lib/allocator/ob_malloc.h"
+#include "lib/ob_errno.h"
+#include "lib/oblog/ob_log.h"
+#include "lib/string/ob_string.h"
+#include "share/backup/ob_backup_io_adapter.h"
+#include "share/backup/ob_backup_struct.h"
+#include "logservice/archiveservice/ob_archive_io.h"
+#include <cstdlib>
+#include <gtest/gtest.h>
+#include <string>
+
+using namespace oceanbase;
+using namespace common;
+using namespace share;
+using namespace archive;
+using namespace std;
+
+TEST(TestObArchiveOverwrite, oss_push_log)
+{
+  int64_t rand_num = rand();
+  ObBackupDest dest;
+  ObBackupIoAdapter util;
+  ObArchiveIO archive_io;
+  const int64_t part_size = 5 * 1024 * 1024L;
+  const int64_t total_size = 10 * 1024 * 1024L;
+  char *data = (char*)ob_malloc(total_size, "ArchiveTest");
+  string path = "oss://antsys-oceanbasebackup/test_archive/test_archive_overwrite/" + to_string(rand_num) + "?host=cn-hangzhou-alipay-b.oss-cdn.aliyun-inc.com&access_id=LTAI4Fdwx9iFgZso4CqyHPs7&" + "access_" + "key=ER51kMn" + "lmu3zXwcxczJ" + "MbYzJIgrY9O";
+
+  EXPECT_EQ(false, NULL == data);
+
+  EXPECT_EQ(0, dest.set(path.c_str()));
+
+  const ObString &uri = dest.get_root_path();
+  const ObBackupStorageInfo *storage_info = dest.get_storage_info();
+
+  // delete file
+  EXPECT_EQ(0, util.del_file(uri, storage_info));
+
+  // scene 1: append & append
+  EXPECT_EQ(0, archive_io.push_log(uri, storage_info, data, part_size, 0, false /*append*/));
+  EXPECT_EQ(0, archive_io.push_log(uri, storage_info, data, total_size, 0, false /*append*/));
+  // delete file
+  EXPECT_EQ(0, util.del_file(uri, storage_info));
+
+  // scene 2: append & put
+  EXPECT_EQ(0, archive_io.push_log(uri, storage_info, data, part_size, 0, false /*append*/));
+  EXPECT_EQ(0, archive_io.push_log(uri, storage_info, data, total_size, 0, true /*put*/));
+  // delete file
+  EXPECT_EQ(0, util.del_file(uri, storage_info));
+
+  // scene 3: put & put
+  EXPECT_EQ(0, archive_io.push_log(uri, storage_info, data, total_size, 0, true /*put*/));
+  EXPECT_EQ(0, archive_io.push_log(uri, storage_info, data, total_size, 0, true /*put*/));
+  // delete file
+  EXPECT_EQ(0, util.del_file(uri, storage_info));
+
+  // scene 4: put & append
+  EXPECT_EQ(0, archive_io.push_log(uri, storage_info, data, total_size, 0, true /*put*/));
+  EXPECT_EQ(0, archive_io.push_log(uri, storage_info, data, part_size, 0, false /*append*/));
+  // delete file
+  EXPECT_EQ(0, util.del_file(uri, storage_info));
+
+  if (NULL != data) {
+    ob_free(data);
+    data = NULL;
+  }
+}
+
+int main(int argc, char **argv)
+{
+  ObLogger::get_logger().set_file_name("test_archive_overwrite.log", true);
+  ObLogger::get_logger().set_log_level("INFO");
+  ::testing::InitGoogleTest(&argc, argv);
+  return RUN_ALL_TESTS();
+}