LogDog: BigTable batching schema.

server/logdog/storage/bigtable/initialize.go

 // Copyright 2015 The Chromium Authors. All rights reserved.
 // Use of this source code is governed by a BSD-style license that can be
 // found in the LICENSE file.
 
 package bigtable
 
 import (
         "fmt"
         "time"
 
@@ skipping 48 matching lines @@
 }
 
 // Initialize sets up a Storage schema in BigTable. If the schema is already
 // set up properly, no action will be taken.
 //
 // If, however, the table or table's schema doesn't exist, Initialize will
 // create and configure it.
 //
 // If nil is returned, the table is ready for use as a Storage via New.
 func Initialize(ctx context.Context, o Options) error {
-        st := New(ctx, o)
+        adminClient, err := o.adminClient(ctx)
+        if err != nil {
+                return err
+        }
+
+        st := newBTStorage(ctx, o, nil, adminClient)
         defer st.Close()
 
-        c, err := st.(*btStorage).getAdminClient()
-        if err != nil {
-                return fmt.Errorf("failed to create admin client: %s", err)
-        }
-
-        exists, err := tableExists(ctx, c, o.LogTable)
+        exists, err := tableExists(ctx, st.adminClient, o.LogTable)
         if err != nil {
                 return fmt.Errorf("failed to test for table: %s", err)
         }
         if !exists {
                 log.Fields{
                         "table": o.LogTable,
                 }.Infof(ctx, "Storage table does not exist. Creating...")
 
-                if err := c.CreateTable(ctx, o.LogTable); err != nil {
+                if err := st.adminClient.CreateTable(ctx, o.LogTable); err != nil {
                         return fmt.Errorf("failed to create table: %s", err)
                 }
 
                 // Wait for the table to exist. BigTable API says this can be delayed from
                 // creation.
-                if err := waitForTable(ctx, c, o.LogTable); err != nil {
+                if err := waitForTable(ctx, st.adminClient, o.LogTable); err != nil {
                         return fmt.Errorf("failed to wait for table to exist: %s", err)
                 }
 
                 log.Fields{
                         "table": o.LogTable,
                 }.Infof(ctx, "Successfully created storage table.")
         }
 
         // Get table info.
-        ti, err := c.TableInfo(ctx, o.LogTable)
+        ti, err := st.adminClient.TableInfo(ctx, o.LogTable)
         if err != nil {
                 return fmt.Errorf("failed to get table info: %s", err)
         }
 
         // The table must have the "log" column family.
         families := stringset.NewFromSlice(ti.Families...)
         if !families.Has(logColumnFamily) {
                 log.Fields{
                         "table":  o.LogTable,
                         "family": logColumnFamily,
                 }.Infof(ctx, "Column family 'log' does not exist. Creating...")
 
                 // Create the logColumnFamily column family.
-                if err := c.CreateColumnFamily(ctx, o.LogTable, logColumnFamily); err != nil {
+                if err := st.adminClient.CreateColumnFamily(ctx, o.LogTable, logColumnFamily); err != nil {
                         return fmt.Errorf("Failed to create 'log' column family: %s", err)
                 }
 
                 log.Fields{
                         "table":  o.LogTable,
                         "family": "log",
                 }.Infof(ctx, "Successfully created 'log' column family.")
         }
 
         cfg := storage.Config{
                 MaxLogAge: DefaultMaxLogAge,
         }
         if err := st.Config(cfg); err != nil {
                 log.WithError(err).Errorf(ctx, "Failed to push default configuration.")
                 return err
         }
 
         return nil
 }