func()

in pkg/ipamd/datastore/data_store.go [370:485]
92 lines of code
18 McCabe index (conditional complexity)

func (ds *DataStore) ReadBackingStore(isv6Enabled bool) error {
	var data CheckpointData

	switch ds.CheckpointMigrationPhase {
	case 1:
		// Phase1: Read from CRI
		ds.log.Infof("Reading ipam state from CRI")
		var ipv4Addr, ipv6Addr string
		sandboxes, err := ds.cri.GetRunningPodSandboxes(ds.log)
		if err != nil {
			return err
		}

		entries := make([]CheckpointEntry, 0, len(sandboxes))
		for _, s := range sandboxes {
			ds.log.Debugf("Adding container ID: %v", s.ID)
			if isv6Enabled {
				ipv6Addr = s.IP
			} else {
				ipv4Addr = s.IP
			}
			entries = append(entries, CheckpointEntry{
				// NB: These Backfill values are also assumed in UnassignPodIPAddress
				IPAMKey: IPAMKey{
					NetworkName: backfillNetworkName,
					ContainerID: s.ID,
					IfName:      backfillNetworkIface,
				},
				IPv4: ipv4Addr,
				IPv6: ipv6Addr,
			})
		}
		data = CheckpointData{
			Version:     CheckpointFormatVersion,
			Allocations: entries,
		}

	case 2:
		// Phase2: Read from checkpoint file
		ds.log.Infof("Reading ipam state from backing store")

		err := ds.backingStore.Restore(&data)
		ds.log.Debugf("backing store restore returned err %v", err)
		if os.IsNotExist(err) {
			// Assume that no file == no containers are
			// currently in use, eg a fresh reboot just
			// cleared everything out.  This is ok, and a
			// no-op.
			return nil
		} else if err != nil {
			return fmt.Errorf("datastore: error reading backing store: %v", err)
		}

		if data.Version != CheckpointFormatVersion {
			return fmt.Errorf("datastore: unknown backing store format (%s != %s) - wrong CNI/ipamd version? (Rebooting this node will restart local pods and probably help)", data.Version, CheckpointFormatVersion)
		}

	default:
		panic(fmt.Sprintf("Unexpected value of checkpointMigrationPhase: %v", ds.CheckpointMigrationPhase))
	}

	ds.lock.Lock()
	defer ds.lock.Unlock()

	for _, allocation := range data.Allocations {
		ipv4Addr := net.ParseIP(allocation.IPv4)
		ipv6Addr := net.ParseIP(allocation.IPv6)
		var ipAddr net.IP
		found := false
	eniloop:
		for _, eni := range ds.eniPool {
			eniCidrs := eni.AvailableIPv4Cidrs
			ipAddr = ipv4Addr
			if isv6Enabled {
				ds.log.Debugf("v6 is enabled")
				eniCidrs = eni.IPv6Cidrs
				ipAddr = ipv6Addr
			}
			for _, cidr := range eniCidrs {
				ds.log.Debugf("Checking if IP: %v belongs to CIDR: %v", ipAddr, cidr.Cidr)
				if cidr.Cidr.Contains(ipAddr) {
					// Found!
					found = true
					if _, ok := cidr.IPAddresses[ipAddr.String()]; ok {
						return errors.New(IPAlreadyInStoreError)
					}
					addr := &AddressInfo{Address: ipAddr.String()}
					cidr.IPAddresses[ipAddr.String()] = addr
					ds.assignPodIPAddressUnsafe(allocation.IPAMKey, eni, addr)
					ds.log.Debugf("Recovered %s => %s/%s", allocation.IPAMKey, eni.ID, addr.Address)
					//Update prometheus for ips per cidr
					//Secondary IP mode will have /32:1 and Prefix mode will have /28:<number of /32s>
					ipsPerCidr.With(prometheus.Labels{"cidr": cidr.Cidr.String()}).Inc()
					break eniloop
				}
			}
		}
		if !found {
			ds.log.Infof("datastore: Sandbox %s uses unknown IP Address %s - presuming stale/dead",
				allocation.IPAMKey, ipAddr.String())
		}
	}

	if ds.CheckpointMigrationPhase == 1 {
		// For phase1: write whatever we just read above from
		// CRI to backingstore immediately - just in case we
		// _never_ see an add/del request before we upgrade to
		// phase2.
		if err := ds.writeBackingStoreUnsafe(); err != nil {
			return err
		}
	}

	ds.log.Debugf("Completed ipam state recovery")
	return nil
}